Files
ems/backend/tests/test_planning_dispatch_milp.py
Dusan Vojacek b46da6b2dc Revert "a dalsi fix"
This reverts commit 7036bcfdb8.
2026-05-25 01:00:00 +02:00

3249 lines
120 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""MILP dispatch: dvouúrovňové SoC a záporná nákupní cena (bez DB)."""
from __future__ import annotations
import unittest
from datetime import datetime, timedelta, timezone
from types import SimpleNamespace
from services.planning_engine import (
DispatchResult,
PlanningSlot,
_dynamic_arb_floor_wh_series,
_dispatch_result_comparison,
_pre_neg_peak_sell_idx,
_prague_hour,
_prewindow_deferral_slots,
_slots_until_buy_le_threshold,
_slots_until_sell_lt,
_soc_panel_min_wh_series,
solve_dispatch,
solve_dispatch_two_pass,
)
def _slot(
*,
load: int = 2000,
buy: float = 3.0,
sell: float = 3.0,
pv_a: int = 0,
pv_b: int = 0,
) -> PlanningSlot:
return PlanningSlot(
interval_start=datetime(2026, 4, 3, 12, 0, tzinfo=timezone.utc),
buy_price=buy,
sell_price=sell,
pv_a_forecast_w=pv_a,
pv_b_forecast_w=pv_b,
load_baseline_w=load,
ev1_connected=False,
ev2_connected=False,
is_predicted_price=False,
)
def _battery(
*,
uc_wh: float = 100_000.0,
min_pct: float = 10.0,
arb_pct: float = 20.0,
max_pct: float = 95.0,
terminal_soc_value_factor: float = 0.9,
) -> SimpleNamespace:
uc = uc_wh
min_wh = min_pct / 100.0 * uc
arb_wh = arb_pct / 100.0 * uc
return SimpleNamespace(
usable_capacity_wh=uc,
min_soc_wh=min_wh,
arb_floor_wh=arb_wh,
reserve_soc_wh=arb_wh,
soc_max_wh=max_pct / 100.0 * uc,
charge_efficiency=0.95,
discharge_efficiency=0.95,
degradation_cost_czk_kwh=0.15,
max_charge_power_w=10_000,
max_discharge_power_w=10_000,
planner_terminal_soc_value_factor=terminal_soc_value_factor,
)
class SlotsUntilSellNegativeTests(unittest.TestCase):
def test_slots_until_first_negative_sell(self) -> None:
base = datetime(2026, 4, 3, 0, 0, tzinfo=timezone.utc)
slots: list[PlanningSlot] = []
for i in range(10):
slots.append(
PlanningSlot(
interval_start=base + timedelta(minutes=15 * i),
buy_price=1.0,
sell_price=2.0 if i < 4 else -0.5,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=500,
ev1_connected=False,
ev2_connected=False,
)
)
dist = _slots_until_sell_lt(slots, 0.0)
self.assertEqual(dist[0], 4)
self.assertEqual(dist[3], 1)
self.assertEqual(dist[4], 0)
def test_prewindow_deferral_prefers_sell_anchor(self) -> None:
"""Když existuje záporný prodej, kotva je vzdálenost k němu, ne k extrémnímu buy."""
base = datetime(2026, 4, 3, 0, 0, tzinfo=timezone.utc)
slots: list[PlanningSlot] = []
for i in range(8):
slots.append(
PlanningSlot(
interval_start=base + timedelta(minutes=15 * i),
buy_price=-50.0,
sell_price=1.0 if i < 2 else -0.1,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=500,
ev1_connected=False,
ev2_connected=False,
)
)
adv = _prewindow_deferral_slots(slots, -2.0)
self.assertEqual(adv[0], 2)
def test_prewindow_deferral_falls_back_to_buy_when_no_negative_sell(self) -> None:
base = datetime(2026, 4, 3, 0, 0, tzinfo=timezone.utc)
slots: list[PlanningSlot] = []
for i in range(10):
slots.append(
PlanningSlot(
interval_start=base + timedelta(minutes=15 * i),
buy_price=3.0 if i < 7 else -10.0,
sell_price=2.0,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=500,
ev1_connected=False,
ev2_connected=False,
)
)
adv = _prewindow_deferral_slots(slots, -2.0)
self.assertEqual(adv[0], 7)
class SlotsUntilBuyExtremeTests(unittest.TestCase):
def test_slots_until_first_extreme(self) -> None:
base = datetime(2026, 4, 3, 0, 0, tzinfo=timezone.utc)
slots: list[PlanningSlot] = []
for i in range(10):
slots.append(
PlanningSlot(
interval_start=base + timedelta(minutes=15 * i),
buy_price=1.0,
sell_price=1.0,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=500,
ev1_connected=False,
ev2_connected=False,
)
)
slots[-1] = PlanningSlot(
interval_start=slots[-1].interval_start,
buy_price=-10.0,
sell_price=0.0,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=500,
ev1_connected=False,
ev2_connected=False,
)
dist = _slots_until_buy_le_threshold(slots, -2.0)
self.assertEqual(dist[0], 9)
self.assertEqual(dist[8], 1)
self.assertEqual(dist[9], 0)
def test_prewindow_clamps_relaxed_floor_until_close(self) -> None:
sm = [5000.0] * 10
dist = [9, 8, 7, 6, 5, 4, 3, 2, 1, 0] # obecná kotva (sell nebo buy)
panel = _soc_panel_min_wh_series(sm, dist, 10_000.0, 20_000.0, 2)
self.assertEqual(panel[0], 20_000.0)
self.assertEqual(panel[6], 20_000.0)
self.assertEqual(panel[7], 5000.0)
self.assertEqual(panel[9], 5000.0)
class DynamicArbFloorTests(unittest.TestCase):
def test_more_pv_ahead_lowers_floor(self) -> None:
"""Čím víc FVE ve lookahead, tím nižší ekonomická podlaha v prvním slotu."""
min_w = 1_000.0
base_w = 2_000.0
uc = 10_000.0
s0 = _slot()
s_low_pv = replace_slot(s0, pv_a=100, pv_b=0)
s_high_pv = replace_slot(s0, pv_a=50_000, pv_b=0)
ser_low = _dynamic_arb_floor_wh_series([s_low_pv] * 40, min_w, base_w, uc)
ser_high = _dynamic_arb_floor_wh_series([s_high_pv] * 40, min_w, base_w, uc)
self.assertLess(ser_high[0], ser_low[0])
self.assertGreaterEqual(ser_low[0], min_w)
self.assertLessEqual(ser_low[0], base_w)
def replace_slot(
s: PlanningSlot,
*,
pv_a: int | None = None,
pv_b: int | None = None,
load: int | None = None,
) -> PlanningSlot:
return PlanningSlot(
interval_start=s.interval_start,
buy_price=s.buy_price,
sell_price=s.sell_price,
pv_a_forecast_w=pv_a if pv_a is not None else s.pv_a_forecast_w,
pv_b_forecast_w=pv_b if pv_b is not None else s.pv_b_forecast_w,
load_baseline_w=load if load is not None else s.load_baseline_w,
ev1_connected=s.ev1_connected,
ev2_connected=s.ev2_connected,
is_predicted_price=s.is_predicted_price,
)
class PlanningDispatchMilpTests(unittest.TestCase):
def test_dispatch_result_comparison_marks_changed_slots(self) -> None:
dt = datetime(2026, 4, 3, 12, 0, tzinfo=timezone.utc)
active = [
DispatchResult(
interval_start=dt,
battery_setpoint_w=1000,
battery_soc_target=50.0,
grid_setpoint_w=0,
export_limit_w=0,
export_mode="NONE",
deye_physical_mode="PASSIVE",
deye_gen_cutoff_enabled=False,
ev1_setpoint_w=None,
ev2_setpoint_w=None,
ev1_via_bat_w=0,
ev2_via_bat_w=0,
heat_pump_enabled=False,
heat_pump_setpoint_w=0,
pv_a_curtailed_w=0,
expected_cost_czk=1.0,
effective_buy_price=1.0,
effective_sell_price=1.0,
is_predicted_price=False,
cashflow_czk=1.0,
battery_arbitrage_czk=0.0,
penalty_czk=0.0,
green_bonus_czk=0.0,
)
]
peer = [
DispatchResult(
interval_start=dt,
battery_setpoint_w=2000,
battery_soc_target=55.0,
grid_setpoint_w=-1000,
export_limit_w=1000,
export_mode="PV_SURPLUS",
deye_physical_mode="SELL",
deye_gen_cutoff_enabled=True,
ev1_setpoint_w=None,
ev2_setpoint_w=None,
ev1_via_bat_w=0,
ev2_via_bat_w=0,
heat_pump_enabled=False,
heat_pump_setpoint_w=0,
pv_a_curtailed_w=200,
expected_cost_czk=2.0,
effective_buy_price=1.0,
effective_sell_price=1.0,
is_predicted_price=False,
cashflow_czk=2.0,
battery_arbitrage_czk=0.0,
penalty_czk=0.0,
green_bonus_czk=0.0,
)
]
cmp = _dispatch_result_comparison(active, 10, "v1", peer, 12, "v2")
self.assertEqual(cmp["active"]["planner_version"], "v1")
self.assertEqual(cmp["peer"]["planner_version"], "v2")
self.assertEqual(cmp["diff"]["changed_slots"], 1)
self.assertEqual(len(cmp["slot_diffs"]), 1)
def test_planner_version_is_recorded_in_snapshot(self) -> None:
slots = [_slot(load=500, buy=1.0, sell=1.0, pv_a=0, pv_b=0) for _ in range(2)]
battery = _battery()
hp = SimpleNamespace(rated_heating_power_w=0, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0)
grid = SimpleNamespace(max_import_power_w=20_000, max_export_power_w=20_000)
vehicles = [
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
]
results, _ms, snap = solve_dispatch(
slots,
battery,
hp,
grid,
[],
vehicles,
current_soc_wh=0.5 * battery.usable_capacity_wh,
current_tuv_temp_c=50.0,
planner_version="v2",
)
self.assertEqual(len(results), 2)
self.assertEqual(snap["inputs"]["planner_version"], "v2")
def test_neg_sell_with_future_neg_buy_prefers_curtail_pv_a_over_export(self) -> None:
"""
Když:
- aktuální slot má sell < 0 (export je náklad),
- v horizontu existuje budoucí buy < 0,
- a zároveň existuje PV B (necurtailable) někde v horizontu,
solver preferuje curtail PV A (ca) místo placeného exportu ge.
"""
slots = [
_slot(load=0, buy=3.0, sell=-0.1, pv_a=5000, pv_b=0),
_slot(load=0, buy=-10.0, sell=1.0, pv_a=0, pv_b=5000),
]
battery = _battery(uc_wh=50_000.0)
hp = SimpleNamespace(
rated_heating_power_w=0,
tuv_min_temp_c=45.0,
tuv_target_temp_c=55.0,
)
grid = SimpleNamespace(max_import_power_w=20_000, max_export_power_w=20_000)
vehicles = [
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
]
soc0 = 0.50 * battery.usable_capacity_wh
results, _ms, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
soc0,
50.0,
tuv_delta_stats=None,
operating_mode="AUTO",
)
self.assertEqual(len(results), 2)
# Slot 0: záporný sell — žádný export FVE do sítě (LP guard sell < acquisition).
self.assertNotEqual(results[0].export_mode, "PV_SURPLUS")
self.assertNotEqual(results[0].export_mode, "PV_SURPLUS")
def test_pv_surplus_export_uses_hard_export_cap(self) -> None:
slots = [
PlanningSlot(
interval_start=datetime(2026, 4, 3, 12, 0, tzinfo=timezone.utc),
buy_price=3.0,
sell_price=2.5,
pv_a_forecast_w=20_000,
pv_b_forecast_w=0,
load_baseline_w=0,
ev1_connected=False,
ev2_connected=False,
is_predicted_price=False,
allow_charge=False,
allow_discharge_export=False,
),
]
battery = _battery()
hp = SimpleNamespace(
rated_heating_power_w=0,
tuv_min_temp_c=45.0,
tuv_target_temp_c=55.0,
)
grid = SimpleNamespace(max_import_power_w=20_000, max_export_power_w=13_500)
vehicles = [
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
]
soc0 = battery.soc_max_wh
results, _ms, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
soc0,
50.0,
tuv_delta_stats=None,
operating_mode="AUTO",
)
self.assertEqual(len(results), 1)
self.assertEqual(results[0].export_mode, "PV_SURPLUS")
self.assertEqual(results[0].export_limit_w, 13_500)
self.assertGreater(results[0].pv_a_curtailed_w, 0)
def test_two_tier_soc_solves_optimal(self) -> None:
slots = [_slot()]
battery = _battery()
hp = SimpleNamespace(
rated_heating_power_w=0,
tuv_min_temp_c=45.0,
tuv_target_temp_c=55.0,
)
grid = SimpleNamespace(max_import_power_w=15_000, max_export_power_w=15_000)
vehicles = [
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
]
soc0 = 0.15 * battery.usable_capacity_wh
results, ms, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
soc0,
50.0,
tuv_delta_stats=None,
operating_mode="AUTO",
)
self.assertGreaterEqual(ms, 0)
self.assertEqual(len(results), 1)
def test_deep_discharge_allows_covering_load_only(self) -> None:
slots = [
_slot(load=3000, buy=1.0, sell=6.0, pv_a=0, pv_b=0),
_slot(load=3000, buy=1.0, sell=6.0, pv_a=0, pv_b=0),
]
battery = _battery(uc_wh=50_000.0)
hp = SimpleNamespace(
rated_heating_power_w=0,
tuv_min_temp_c=45.0,
tuv_target_temp_c=55.0,
)
grid = SimpleNamespace(max_import_power_w=20_000, max_export_power_w=20_000)
vehicles = [
SimpleNamespace(
max_charge_power_w=11_000,
battery_capacity_kwh=50.0,
default_target_soc_pct=80.0,
),
SimpleNamespace(
max_charge_power_w=11_000,
battery_capacity_kwh=50.0,
default_target_soc_pct=80.0,
),
]
soc0 = 0.12 * battery.usable_capacity_wh
results, _ms, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
soc0,
50.0,
tuv_delta_stats=None,
operating_mode="AUTO",
)
self.assertEqual(len(results), 2)
def test_negative_buy_price_allows_import_for_baseline(self) -> None:
slots = [_slot(load=6000, buy=-0.5, sell=2.0)]
battery = _battery()
hp = SimpleNamespace(
rated_heating_power_w=8000,
tuv_min_temp_c=45.0,
tuv_target_temp_c=55.0,
)
grid = SimpleNamespace(max_import_power_w=25_000, max_export_power_w=15_000)
vehicles = [
SimpleNamespace(
max_charge_power_w=11_000,
battery_capacity_kwh=50.0,
default_target_soc_pct=80.0,
),
SimpleNamespace(
max_charge_power_w=11_000,
battery_capacity_kwh=50.0,
default_target_soc_pct=80.0,
),
]
soc0 = 0.5 * battery.usable_capacity_wh
results, _ms, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
soc0,
50.0,
tuv_delta_stats=None,
operating_mode="AUTO",
)
self.assertGreaterEqual(results[0].grid_setpoint_w, 0)
def test_export_implies_end_soc_at_least_reserve(self) -> None:
"""Bez arbitrážní relaxace: při ge >= 1 W musí koncové soc[t] >= arb_base_wh (rezerva z DB)."""
slots = [
_slot(load=500, buy=2.0, sell=8.0, pv_a=0, pv_b=0),
_slot(load=500, buy=2.0, sell=8.0, pv_a=0, pv_b=0),
]
battery = _battery(uc_wh=100_000.0, min_pct=10.0, arb_pct=20.0)
hp = SimpleNamespace(
rated_heating_power_w=0,
tuv_min_temp_c=45.0,
tuv_target_temp_c=55.0,
)
grid = SimpleNamespace(max_import_power_w=50_000, max_export_power_w=50_000)
vehicles = [
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
]
soc0 = 0.22 * battery.usable_capacity_wh
results, _ms, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
soc0,
50.0,
tuv_delta_stats=None,
operating_mode="AUTO",
)
reserve_pct = 20.0
for r in results:
if r.grid_setpoint_w < 0:
self.assertGreaterEqual(
r.battery_soc_target,
reserve_pct - 0.2,
msg="export slot must end at or above reserve SoC",
)
def test_export_before_extreme_negative_buy_can_end_below_reserve(self) -> None:
"""
Při relaxovaném soc_min (záporný buy v lookahead) smí významný export skončit u planner floor,
ne u provozní rezervy — jinak nejde ráno vypustit do sítě a nachystat kapacitu před levným nákupem.
"""
base = datetime(2026, 4, 3, 6, 0, tzinfo=timezone.utc)
s0 = PlanningSlot(
interval_start=base,
buy_price=2.5,
sell_price=2.0,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=400,
ev1_connected=False,
ev2_connected=False,
is_predicted_price=False,
allow_charge=True,
allow_discharge_export=True,
)
s1 = PlanningSlot(
interval_start=base + timedelta(minutes=15),
buy_price=-12.0,
sell_price=-0.5,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=400,
ev1_connected=False,
ev2_connected=False,
is_predicted_price=False,
allow_charge=True,
allow_discharge_export=True,
)
slots = [s0, s1]
battery = _battery(uc_wh=10_000.0, min_pct=10.0, arb_pct=20.0)
battery.planner_extreme_buy_threshold_czk_kwh = -2.0
battery.planner_discharge_floor_percent = 5.0
battery.max_charge_power_w = 50_000
battery.max_discharge_power_w = 50_000
hp = SimpleNamespace(
rated_heating_power_w=0,
tuv_min_temp_c=45.0,
tuv_target_temp_c=55.0,
)
grid = SimpleNamespace(max_import_power_w=50_000, max_export_power_w=50_000)
vehicles = [
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
]
soc0 = 0.88 * battery.usable_capacity_wh
results, _ms, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
soc0,
50.0,
tuv_delta_stats=None,
operating_mode="AUTO",
)
self.assertEqual(len(results), 2)
if results[0].grid_setpoint_w < 0:
self.assertLess(
results[0].battery_soc_target,
22.0,
msg="with relaxed soc_min, morning export should finish below reserve %",
)
def test_negative_sell_forbids_battery_export_arbitrage(self) -> None:
"""
Pokud sell < 0, solver nesmí vybíjet baterii do sítě pro arbitráž (dump musí proběhnout předtím).
V okně sell<0 smí export vzniknout jen z přebytku FVE; zde ale FVE=0, takže očekáváme grid_setpoint>=0.
"""
base = datetime(2026, 4, 3, 6, 0, tzinfo=timezone.utc)
s0 = PlanningSlot(
interval_start=base,
buy_price=2.0,
sell_price=2.0,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=0,
ev1_connected=False,
ev2_connected=False,
is_predicted_price=False,
allow_charge=True,
allow_discharge_export=True,
)
s1 = PlanningSlot(
interval_start=base + timedelta(minutes=15),
buy_price=2.0,
sell_price=-0.5,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=0,
ev1_connected=False,
ev2_connected=False,
is_predicted_price=False,
allow_charge=True,
allow_discharge_export=True,
)
s2 = PlanningSlot(
interval_start=base + timedelta(minutes=30),
buy_price=-15.0,
sell_price=-1.0,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=0,
ev1_connected=False,
ev2_connected=False,
is_predicted_price=False,
allow_charge=True,
allow_discharge_export=True,
)
slots = [s0, s1, s2]
battery = _battery(uc_wh=10_000.0, min_pct=10.0, arb_pct=20.0)
battery.planner_extreme_buy_threshold_czk_kwh = -2.0
battery.planner_discharge_floor_percent = 5.0
battery.max_charge_power_w = 50_000
battery.max_discharge_power_w = 50_000
hp = SimpleNamespace(
rated_heating_power_w=0,
tuv_min_temp_c=45.0,
tuv_target_temp_c=55.0,
)
grid = SimpleNamespace(max_import_power_w=50_000, max_export_power_w=50_000)
vehicles = [
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
]
soc0 = 0.9 * battery.usable_capacity_wh
results, _ms, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
soc0,
50.0,
tuv_delta_stats=None,
operating_mode="AUTO",
)
self.assertEqual(len(results), 3)
# V sell<0 slotu bez FVE a bez zátěže nesmí být export (to by muselo být z baterie).
self.assertGreaterEqual(results[1].grid_setpoint_w, 0)
# A zároveň nesmí být baterie ve výboji (dump musí proběhnout předtím).
self.assertGreaterEqual(results[1].battery_setpoint_w, 0)
def test_anchor_hits_floor_before_first_negative_sell(self) -> None:
"""
Pokud se v horizontu objeví první sell<0 a současně existuje planner floor (relaxace),
solver má skončit už v předchozím slotu u planner floor (cca 5 %), ne na ~15 %.
"""
base = datetime(2026, 4, 3, 6, 0, tzinfo=timezone.utc)
# Slot 0-1: sell >= 0; slot 2: první sell < 0; slot 3: extrémně záporný buy (motivace k bufferu).
slots = [
PlanningSlot(
interval_start=base,
buy_price=3.0,
sell_price=1.0,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=0,
ev1_connected=False,
ev2_connected=False,
allow_charge=True,
allow_discharge_export=True,
),
PlanningSlot(
interval_start=base + timedelta(minutes=15),
buy_price=3.0,
sell_price=0.5,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=0,
ev1_connected=False,
ev2_connected=False,
allow_charge=True,
allow_discharge_export=True,
),
PlanningSlot(
interval_start=base + timedelta(minutes=30),
buy_price=3.0,
sell_price=-0.2,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=0,
ev1_connected=False,
ev2_connected=False,
allow_charge=True,
allow_discharge_export=True,
),
PlanningSlot(
interval_start=base + timedelta(minutes=45),
buy_price=-20.0,
sell_price=-1.0,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=0,
ev1_connected=False,
ev2_connected=False,
allow_charge=True,
allow_discharge_export=True,
),
]
battery = _battery(uc_wh=20_000.0, min_pct=10.0, arb_pct=20.0)
battery.planner_extreme_buy_threshold_czk_kwh = -2.0
battery.planner_discharge_floor_percent = 5.0
battery.max_charge_power_w = 50_000
battery.max_discharge_power_w = 50_000
hp = SimpleNamespace(rated_heating_power_w=0, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0)
grid = SimpleNamespace(max_import_power_w=50_000, max_export_power_w=50_000)
vehicles = [
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
]
soc0 = 0.9 * battery.usable_capacity_wh
results, _ms, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
soc0,
50.0,
tuv_delta_stats=None,
operating_mode="AUTO",
)
peak_t = _pre_neg_peak_sell_idx(slots, 2)
self.assertIsNotNone(peak_t)
self.assertLess(
results[peak_t].grid_setpoint_w,
-500,
msg="ranní peak: export baterie/FVE před sell<0",
)
def test_anchor_uses_planner_floor_even_without_extreme_buy(self) -> None:
"""
Regrese: pokud v horizontu není buy <= threshold (soc_min_series by se nerelaxovala),
kotva před sell<0 má stejně mířit na planner floor (5 %), ne na base min SoC.
"""
base = datetime(2026, 4, 3, 6, 0, tzinfo=timezone.utc)
slots = [
PlanningSlot(
interval_start=base,
buy_price=3.0,
sell_price=3.06,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=0,
ev1_connected=False,
ev2_connected=False,
allow_charge=True,
allow_discharge_export=True,
),
PlanningSlot(
interval_start=base + timedelta(minutes=15),
buy_price=3.0,
sell_price=2.0,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=0,
ev1_connected=False,
ev2_connected=False,
allow_charge=True,
allow_discharge_export=True,
),
PlanningSlot(
interval_start=base + timedelta(minutes=30),
buy_price=3.0,
sell_price=-0.2,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=0,
ev1_connected=False,
ev2_connected=False,
allow_charge=True,
allow_discharge_export=True,
),
]
battery = _battery(uc_wh=20_000.0, min_pct=12.0, arb_pct=20.0)
battery.planner_extreme_buy_threshold_czk_kwh = -2.0
battery.planner_discharge_floor_percent = 5.0
battery.max_charge_power_w = 50_000
battery.max_discharge_power_w = 50_000
hp = SimpleNamespace(rated_heating_power_w=0, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0)
grid = SimpleNamespace(max_import_power_w=50_000, max_export_power_w=50_000)
vehicles = [
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
]
soc0 = 0.9 * battery.usable_capacity_wh
results, _ms, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
soc0,
50.0,
tuv_delta_stats=None,
operating_mode="AUTO",
)
self.assertLess(
results[0].grid_setpoint_w,
-1_000,
msg="morning peak slot should export before first negative sell",
)
def test_grid_import_soft_cap_penalizes_breaker_overdraw(self) -> None:
"""
Soft cap: solver může nominálně překročit breaker, ale jen pokud se to vyplatí.
Při běžné (nezáporné) nákupní ceně by měl držet import <= breaker.
"""
slots = [_slot(load=3700, buy=0.4, sell=-0.3, pv_a=0, pv_b=1500)]
battery = _battery(uc_wh=64_000.0, min_pct=12.0, arb_pct=20.0)
battery.max_charge_power_w = 18_000
battery.max_discharge_power_w = 18_000
hp = SimpleNamespace(
rated_heating_power_w=0,
tuv_min_temp_c=45.0,
tuv_target_temp_c=55.0,
)
grid = SimpleNamespace(max_import_power_w=17_000, max_export_power_w=13_500)
vehicles = [
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
]
soc0 = 0.55 * battery.usable_capacity_wh
results, _ms, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
soc0,
50.0,
tuv_delta_stats=None,
operating_mode="AUTO",
)
self.assertEqual(len(results), 1)
self.assertLessEqual(
results[0].grid_setpoint_w,
grid.max_import_power_w,
msg="soft cap: for normal buy price, planned grid import should not exceed breaker",
)
def test_grid_import_soft_cap_allows_overdraw_when_extremely_negative(self) -> None:
"""
Regrese: při extrémně záporné nákupní ceně může solver překročit breaker (za cenu penalizace),
aby stihl krátké okno nabíjení. Překročení nesmí být 'zadarmo' (kontrolujeme alespoň, že existuje).
"""
# Dvouslotový scénář: v 1. slotu extrémně záporná cena, ve 2. slotu drahá.
# Terminal SoC kotva pak nepenalizuje držení energie (průměrná buy je ~0) a solver má motivaci
# v 1. slotu nabít na max, i kdyby to znamenalo malé překročení breakeru.
s0 = _slot(load=0, buy=-20.0, sell=-0.3, pv_a=0, pv_b=0)
s1 = replace_slot(s0, load=0)
s1 = PlanningSlot(
interval_start=s0.interval_start + timedelta(minutes=15),
buy_price=20.0,
sell_price=-0.3,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=0,
ev1_connected=False,
ev2_connected=False,
is_predicted_price=False,
)
slots = [s0, s1]
battery = _battery(uc_wh=64_000.0, min_pct=12.0, arb_pct=20.0)
battery.max_charge_power_w = 18_000
battery.max_discharge_power_w = 18_000
hp = SimpleNamespace(
rated_heating_power_w=0,
tuv_min_temp_c=45.0,
tuv_target_temp_c=55.0,
)
grid = SimpleNamespace(max_import_power_w=17_000, max_export_power_w=13_500)
vehicles = [
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
]
soc0 = 0.15 * battery.usable_capacity_wh
results, _ms, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
soc0,
50.0,
tuv_delta_stats=None,
operating_mode="AUTO",
)
self.assertEqual(len(results), 2)
self.assertGreater(
results[0].battery_setpoint_w + max(0, results[0].grid_setpoint_w),
2_000,
msg="záporný buy má vést k nabíjení baterie nebo importu",
)
def test_block_export_on_negative_sell_no_grid_export_pv_surplus(self) -> None:
"""site_grid_connection.block_export_on_negative_sell → ge=0 při sell<0."""
slots = [
PlanningSlot(
interval_start=datetime(2026, 4, 3, 12, 0, tzinfo=timezone.utc),
buy_price=5.25,
sell_price=-0.5,
pv_a_forecast_w=7000,
pv_b_forecast_w=0,
load_baseline_w=500,
ev1_connected=False,
ev2_connected=False,
is_predicted_price=False,
allow_charge=True,
allow_discharge_export=False,
)
]
battery = _battery(uc_wh=20_000.0, arb_pct=15.0, max_pct=95.0)
hp = SimpleNamespace(
rated_heating_power_w=0,
tuv_min_temp_c=45.0,
tuv_target_temp_c=55.0,
)
grid = SimpleNamespace(
max_import_power_w=17_000,
max_export_power_w=8000,
block_export_on_negative_sell=True,
)
vehicles = [
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
]
soc0 = 0.34 * battery.usable_capacity_wh
results, _ms, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
soc0,
50.0,
tuv_delta_stats=None,
operating_mode="AUTO",
)
self.assertEqual(len(results), 1)
self.assertGreaterEqual(results[0].grid_setpoint_w, 0, "no grid export")
self.assertGreater(results[0].battery_setpoint_w, 0, "surplus PV should charge")
class NegativeSellPvChargeTests(unittest.TestCase):
"""BA81: při sell<0 a velké FVE A má jít výkon do baterie, ne do curtailment."""
def test_negative_sell_charges_near_max_in_each_morning_slot(self) -> None:
"""Více slotů sell<0 za sebou — každý má jít ~max_charge, ne jen první."""
base = datetime(2026, 5, 24, 6, 0, tzinfo=timezone.utc)
slots: list[PlanningSlot] = []
for i in range(6):
slots.append(
PlanningSlot(
interval_start=base + timedelta(minutes=15 * i),
buy_price=3.088,
sell_price=-0.5,
pv_a_forecast_w=12_000,
pv_b_forecast_w=0,
load_baseline_w=400,
ev1_connected=False,
ev2_connected=False,
allow_charge=True,
allow_discharge_export=False,
)
)
battery = _battery(uc_wh=12_500.0, terminal_soc_value_factor=0.2)
battery.max_charge_power_w = 6_250
battery.max_discharge_power_w = 6_250
hp = SimpleNamespace(
rated_heating_power_w=0,
tuv_min_temp_c=45.0,
tuv_target_temp_c=55.0,
)
grid = SimpleNamespace(
max_import_power_w=17_000,
max_export_power_w=16_000,
block_export_on_negative_sell=False,
)
vehicles = [
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
]
soc0 = 0.30 * battery.usable_capacity_wh
results, _ms, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
soc0,
50.0,
operating_mode="AUTO",
)
high_power = [r.battery_setpoint_w for r in results if r.battery_setpoint_w > 5_500]
self.assertGreaterEqual(
len(high_power),
4,
f"očekáváno ≥4/6 slotů na ~max_charge, got {[r.battery_setpoint_w for r in results]}",
)
def test_negative_sell_prefers_full_pv_charge_over_curtail(self) -> None:
slots = [
PlanningSlot(
interval_start=datetime(2026, 5, 24, 9, 0, tzinfo=timezone.utc),
buy_price=3.088,
sell_price=-0.9,
pv_a_forecast_w=13_500,
pv_b_forecast_w=0,
load_baseline_w=500,
ev1_connected=False,
ev2_connected=False,
allow_charge=True,
allow_discharge_export=False,
)
]
battery = _battery(uc_wh=12_500.0, terminal_soc_value_factor=0.2)
battery.max_charge_power_w = 6_250
battery.max_discharge_power_w = 6_250
hp = SimpleNamespace(
rated_heating_power_w=0,
tuv_min_temp_c=45.0,
tuv_target_temp_c=55.0,
)
grid = SimpleNamespace(
max_import_power_w=17_000,
max_export_power_w=16_000,
block_export_on_negative_sell=False,
)
vehicles = [
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
]
soc0 = 0.33 * battery.usable_capacity_wh
results, _ms, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
soc0,
50.0,
operating_mode="AUTO",
)
r0 = results[0]
self.assertGreater(
r0.battery_setpoint_w,
5_500,
"při sell<0 a PV≈13 kW má baterie nabíjet blízko max_charge (6,25 kW)",
)
# Přebytek nad max_charge jde do curtail (ne ~3 kW nabíjení + 9 kW curtail při plné baterii).
self.assertGreater(
r0.battery_setpoint_w,
r0.pv_a_curtailed_w * 0.5,
"nabíjení má dominovat nad curtailmentem",
)
def test_negative_sell_charges_from_plateau_soc_without_allow_charge_mask(self) -> None:
"""BA81: allow_charge=false z DB nesmí vypnout shortfall — charge_slots z sell<0 + PV."""
base = datetime(2026, 5, 24, 4, 15, tzinfo=timezone.utc)
slots: list[PlanningSlot] = []
for i in range(6):
h = 6 + (i * 15) // 60
m = (i * 15) % 60
hour_f = max(0.0, min(1.0, (h + m / 60.0 - 6.0) / 14.0))
safety = 3750.0 + 2500.0 * hour_f
slots.append(
PlanningSlot(
interval_start=base + timedelta(minutes=15 * i),
buy_price=3.088,
sell_price=-0.3,
pv_a_forecast_w=9000,
pv_b_forecast_w=800,
load_baseline_w=150,
ev1_connected=False,
ev2_connected=False,
allow_charge=False,
allow_discharge_export=False,
safety_soc_target_wh=safety,
is_daytime_pv_surplus_slot=True,
future_sell_opportunity_czk_kwh=3.7,
)
)
battery = _battery(uc_wh=12_500.0, terminal_soc_value_factor=0.2)
battery.max_charge_power_w = 6_250
battery.max_discharge_power_w = 6_250
hp = SimpleNamespace(
rated_heating_power_w=0,
tuv_min_temp_c=45.0,
tuv_target_temp_c=55.0,
)
grid = SimpleNamespace(
max_import_power_w=17_000,
max_export_power_w=16_000,
block_export_on_negative_sell=False,
)
vehicles = [
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
]
soc0 = 0.508 * battery.usable_capacity_wh
results, _ms, snap = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
soc0,
50.0,
operating_mode="AUTO",
)
self.assertEqual(snap.get("planner_build_tag"), "2026-05-27-pre-neg-buy-soc-cap-v14")
self.assertGreater(
results[0].battery_setpoint_w,
5_500,
f"od ~51 % SoC má první neg slot nabíjet max, got {[r.battery_setpoint_w for r in results]}",
)
self.assertGreaterEqual(
max(r.battery_soc_target for r in results),
round(float(battery.soc_max_wh) / battery.usable_capacity_wh * 100, 1) - 0.5,
"neg okno má dobít na planner soc_max, ne ~92 %",
)
def test_fixed_tariff_evening_export_when_sell_above_buy(self) -> None:
"""BA81: sell 3,7 > buy 3,088 musí exportovat (acq 3,61 + 0,3 by dříve blokovalo)."""
slots = [
PlanningSlot(
interval_start=datetime(2026, 5, 24, 17, 0, tzinfo=timezone.utc),
buy_price=3.088,
sell_price=3.75,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=400,
ev1_connected=False,
ev2_connected=False,
allow_charge=False,
allow_discharge_export=True,
charge_acquisition_buy_czk_kwh=3.613,
)
]
battery = _battery(uc_wh=12_500.0, terminal_soc_value_factor=0.0)
battery.max_discharge_power_w = 6_250
battery.planner_daytime_charge_target_enabled = False
hp = SimpleNamespace(
rated_heating_power_w=0,
tuv_min_temp_c=45.0,
tuv_target_temp_c=55.0,
)
grid = SimpleNamespace(
max_import_power_w=17_000,
max_export_power_w=16_000,
block_export_on_negative_sell=False,
purchase_pricing_mode="fixed",
sale_pricing_mode="spot",
)
vehicles = [
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
]
soc0 = 0.95 * battery.usable_capacity_wh
results, _ms, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
soc0,
50.0,
operating_mode="AUTO",
)
r0 = results[0]
export_w = max(0, -r0.grid_setpoint_w) + max(0, -r0.battery_setpoint_w)
self.assertGreater(
export_w,
0,
"kladný sell>buy: alespoň částečný výdej (jednoslotový horizont — plný push až v integračním testu)",
)
def test_fixed_tariff_post_neg_pv_b_full_soc_feasible(self) -> None:
"""BA81: plná baterie + sell<0 + odpoledne pv_b — ge_pv==0 z pv_store dříve dělalo Infeasible."""
slots = [
PlanningSlot(
interval_start=datetime(2026, 5, 24, 6, 0, tzinfo=timezone.utc)
+ timedelta(minutes=15 * i),
buy_price=3.088,
sell_price=-0.8,
pv_a_forecast_w=12_000,
pv_b_forecast_w=0,
load_baseline_w=500,
ev1_connected=False,
ev2_connected=False,
allow_charge=True,
allow_discharge_export=False,
charge_acquisition_buy_czk_kwh=3.61,
)
for i in range(6)
]
slots.append(
PlanningSlot(
interval_start=datetime(2026, 5, 24, 12, 0, tzinfo=timezone.utc),
buy_price=3.088,
sell_price=3.2,
pv_a_forecast_w=0,
pv_b_forecast_w=2_500,
load_baseline_w=500,
ev1_connected=False,
ev2_connected=False,
allow_charge=False,
allow_discharge_export=False,
charge_acquisition_buy_czk_kwh=3.61,
future_sell_opportunity_czk_kwh=3.76,
)
)
battery = _battery(uc_wh=12_500.0, terminal_soc_value_factor=0.0)
battery.max_charge_power_w = 6_250
battery.max_discharge_power_w = 6_250
battery.degradation_cost_czk_kwh = 0.3
battery.planner_daytime_charge_target_enabled = False
hp = SimpleNamespace(
rated_heating_power_w=0,
tuv_min_temp_c=45.0,
tuv_target_temp_c=55.0,
)
grid = SimpleNamespace(
max_import_power_w=17_000,
max_export_power_w=16_000,
block_export_on_negative_sell=False,
)
vehicles = [
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
]
soc0 = 0.95 * battery.usable_capacity_wh
results, _ms, snap = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
soc0,
50.0,
operating_mode="AUTO",
)
self.assertEqual(snap.get("planner_build_tag"), "2026-05-27-pre-neg-buy-soc-cap-v14")
self.assertEqual(len(results), len(slots))
def test_gen_cutoff_full_soc_neg_sell_with_pv_b_feasible(self) -> None:
"""BA81: 100 % SoC + sell<0 + GEN cut-off — dříve ge==0 → Infeasible."""
slots = [
PlanningSlot(
interval_start=datetime(2026, 5, 24, 9, 0, tzinfo=timezone.utc)
+ timedelta(minutes=15 * i),
buy_price=3.088,
sell_price=-1.5,
pv_a_forecast_w=8_000,
pv_b_forecast_w=2_800,
load_baseline_w=500,
ev1_connected=False,
ev2_connected=False,
allow_charge=True,
allow_discharge_export=False,
charge_acquisition_buy_czk_kwh=3.61,
is_daytime_pv_surplus_slot=True,
safety_soc_target_wh=6_250.0,
)
for i in range(8)
]
battery = _battery(uc_wh=12_500.0, terminal_soc_value_factor=0.0)
battery.soc_max_wh = 12_500.0
battery.max_charge_power_w = 6_250
battery.max_discharge_power_w = 6_250
battery.degradation_cost_czk_kwh = 0.3
battery.planner_daytime_charge_target_enabled = True
hp = SimpleNamespace(
rated_heating_power_w=0,
tuv_min_temp_c=0.0,
tuv_target_temp_c=55.0,
)
grid = SimpleNamespace(
max_import_power_w=17_000,
max_export_power_w=16_000,
block_export_on_negative_sell=False,
deye_gen_microinverter_cutoff_enabled=True,
)
vehicles = [
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
]
results, _ms, snap = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
12_500.0,
55.0,
operating_mode="AUTO",
)
self.assertEqual(snap.get("planner_build_tag"), "2026-05-27-pre-neg-buy-soc-cap-v14")
self.assertEqual(len(results), len(slots))
def test_fixed_tariff_neg_sell_no_grid_export(self) -> None:
"""BA81: sell<0 nesmí vést do sítě (záporná výkupní cena) — jen nabíjení/curtail."""
slots = [
PlanningSlot(
interval_start=datetime(2026, 5, 25, 7, 0, tzinfo=timezone.utc)
+ timedelta(minutes=15 * i),
buy_price=3.088,
sell_price=-0.5,
pv_a_forecast_w=10_000,
pv_b_forecast_w=2_500,
load_baseline_w=300,
ev1_connected=False,
ev2_connected=False,
allow_charge=True,
allow_discharge_export=False,
charge_acquisition_buy_czk_kwh=3.61,
)
for i in range(4)
]
battery = _battery(uc_wh=12_500.0, terminal_soc_value_factor=0.0)
battery.soc_max_wh = 12_500.0
battery.max_charge_power_w = 6_250
grid = SimpleNamespace(
max_import_power_w=17_000,
max_export_power_w=16_000,
block_export_on_negative_sell=False,
deye_gen_microinverter_cutoff_enabled=True,
purchase_pricing_mode="fixed",
sale_pricing_mode="spot",
)
hp = SimpleNamespace(
rated_heating_power_w=0,
tuv_min_temp_c=45.0,
tuv_target_temp_c=55.0,
)
vehicles = [
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
]
results, _ms, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
8_000.0,
50.0,
operating_mode="AUTO",
)
for r in results:
self.assertGreaterEqual(r.battery_setpoint_w, 0, "neg sell má nabíjet")
self.assertGreaterEqual(r.grid_setpoint_w, 0, "neg sell bez exportu do sítě")
def test_ba81_fixed_purchase_nt_vt_buy_spread_neg_sell_no_export(self) -> None:
"""BA81: NT/VT buy v horizontu (rozptyl >0,25) — záporný sell stále bez exportu."""
slots = [
PlanningSlot(
interval_start=datetime(2026, 5, 25, 7, 0, tzinfo=timezone.utc)
+ timedelta(minutes=15 * i),
buy_price=3.088 if i % 2 == 0 else 4.086,
sell_price=-0.5,
pv_a_forecast_w=10_000,
pv_b_forecast_w=2_500,
load_baseline_w=300,
ev1_connected=False,
ev2_connected=False,
allow_charge=True,
allow_discharge_export=False,
charge_acquisition_buy_czk_kwh=3.61,
)
for i in range(4)
]
self.assertGreater(
max(s.buy_price for s in slots) - min(s.buy_price for s in slots),
0.25,
)
battery = _battery(uc_wh=12_500.0, terminal_soc_value_factor=0.0)
battery.soc_max_wh = 12_500.0
battery.max_charge_power_w = 6_250
grid = SimpleNamespace(
max_import_power_w=17_000,
max_export_power_w=16_000,
block_export_on_negative_sell=False,
deye_gen_microinverter_cutoff_enabled=True,
purchase_pricing_mode="fixed",
sale_pricing_mode="spot",
)
hp = SimpleNamespace(
rated_heating_power_w=0,
tuv_min_temp_c=45.0,
tuv_target_temp_c=55.0,
)
vehicles = [
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
]
results, _ms, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
8_000.0,
50.0,
operating_mode="AUTO",
)
for r in results:
self.assertGreaterEqual(r.grid_setpoint_w, 0)
class AutoPvSurplusExportTests(unittest.TestCase):
"""Plná baterie + vysoká FVE: export přebytku (ge_pv), ne curtailment, bez SELL."""
def test_pv_surplus_exports_when_battery_export_disallowed(self) -> None:
slots = [
PlanningSlot(
interval_start=datetime(2026, 5, 17, 10, 0, tzinfo=timezone.utc),
buy_price=1.20,
sell_price=0.80,
pv_a_forecast_w=0,
pv_b_forecast_w=12_000,
load_baseline_w=2000,
ev1_connected=False,
ev2_connected=False,
is_predicted_price=False,
allow_charge=False,
allow_discharge_export=False,
),
]
battery = _battery(uc_wh=20_000.0, min_pct=10.0, arb_pct=20.0)
battery.planner_terminal_soc_value_factor = 0.0
battery.planner_daytime_charge_target_enabled = False
hp = SimpleNamespace(
rated_heating_power_w=0,
tuv_min_temp_c=45.0,
tuv_target_temp_c=55.0,
)
grid = SimpleNamespace(max_import_power_w=17_000, max_export_power_w=8000)
vehicles = [
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
]
soc0 = 0.95 * battery.soc_max_wh
results, _, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
soc0,
50.0,
tuv_delta_stats=None,
operating_mode="AUTO",
)
self.assertLess(results[0].grid_setpoint_w, 0, "PV surplus should export to grid")
self.assertEqual(results[0].deye_physical_mode, "PASSIVE")
self.assertEqual(results[0].export_mode, "PV_SURPLUS")
self.assertLess(results[0].pv_a_curtailed_w, 5000, "should not curtail all PV")
class AutoPassiveSelfConsumptionTests(unittest.TestCase):
"""AUTO bez allow_discharge_export: vlastní spotřeba, ne export do sítě."""
def test_expensive_slot_prefers_battery_over_grid_import(self) -> None:
base = datetime(2026, 5, 16, 22, 0, tzinfo=timezone.utc)
slots = [
PlanningSlot(
interval_start=base,
buy_price=4.80,
sell_price=2.90,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=1200,
ev1_connected=False,
ev2_connected=False,
is_predicted_price=False,
allow_charge=False,
allow_discharge_export=False,
),
PlanningSlot(
interval_start=base + timedelta(minutes=15),
buy_price=0.50,
sell_price=-0.20,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=1200,
ev1_connected=False,
ev2_connected=False,
is_predicted_price=False,
allow_charge=True,
allow_discharge_export=False,
),
]
battery = _battery(uc_wh=20_000.0, min_pct=10.0, arb_pct=20.0)
battery.planner_terminal_soc_value_factor = 0.0
battery.planner_daytime_charge_target_enabled = False
hp = SimpleNamespace(
rated_heating_power_w=0,
tuv_min_temp_c=45.0,
tuv_target_temp_c=55.0,
)
grid = SimpleNamespace(max_import_power_w=17_000, max_export_power_w=8000)
vehicles = [
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
]
soc0 = 0.23 * battery.usable_capacity_wh
results, _, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
soc0,
50.0,
tuv_delta_stats=None,
operating_mode="AUTO",
)
self.assertLess(
results[0].battery_setpoint_w,
0,
msg="expensive slot should discharge for self-consumption before cheap charge",
)
self.assertLessEqual(
results[0].grid_setpoint_w,
0,
msg="expensive slot: baseline load ze baterie, ne import ze sítě",
)
self.assertEqual(results[0].deye_physical_mode, "PASSIVE")
def test_fixed_tariff_expensive_slot_discharges_not_grid_load(self) -> None:
"""KV1 typ: konstantní buy — porovnání vůči charge_acquisition, ne min(buy)."""
slots = [
PlanningSlot(
interval_start=datetime(2026, 5, 21, 22, 0, tzinfo=timezone.utc),
buy_price=6.35,
sell_price=2.0,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=320,
ev1_connected=False,
ev2_connected=False,
allow_charge=False,
allow_discharge_export=False,
charge_acquisition_buy_czk_kwh=0.55,
)
]
battery = _battery(uc_wh=20_000.0, min_pct=10.0, arb_pct=20.0)
battery.planner_terminal_soc_value_factor = 0.0
hp = SimpleNamespace(rated_heating_power_w=0, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0)
grid = SimpleNamespace(max_import_power_w=17_000, max_export_power_w=8000)
vehicles = [
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
]
soc0 = 0.4 * battery.usable_capacity_wh
results, _, _ = solve_dispatch(
slots, battery, hp, grid, [None, None], vehicles, soc0, 50.0, operating_mode="AUTO"
)
self.assertLessEqual(results[0].grid_setpoint_w, 0)
self.assertLess(results[0].battery_setpoint_w, -100)
def test_expensive_slot_uses_hp_variable_not_rated(self) -> None:
"""Regrese: bd+pv_ld >= load+hp[t], ne load+hp_rated (jinak Infeasible bez PV)."""
slots = [
PlanningSlot(
interval_start=datetime(2026, 5, 22, 20, 0, tzinfo=timezone.utc),
buy_price=3.0,
sell_price=2.0,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=1961,
ev1_connected=False,
ev2_connected=False,
allow_charge=False,
allow_discharge_export=False,
charge_acquisition_buy_czk_kwh=0.52,
),
PlanningSlot(
interval_start=datetime(2026, 5, 22, 20, 15, tzinfo=timezone.utc),
buy_price=-5.0,
sell_price=2.0,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=1961,
ev1_connected=False,
ev2_connected=False,
allow_charge=False,
allow_discharge_export=False,
charge_acquisition_buy_czk_kwh=0.52,
),
]
battery = _battery(uc_wh=64_000.0, min_pct=12.0, arb_pct=20.0)
battery.planner_terminal_soc_value_factor = 0.0
hp = SimpleNamespace(rated_heating_power_w=3500, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0)
grid = SimpleNamespace(max_import_power_w=17_000, max_export_power_w=13_500)
vehicles = [
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
]
results, _, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
20_000.0,
50.0,
operating_mode="AUTO",
)
self.assertEqual(len(results), 2)
def test_negative_buy_in_horizon_does_not_block_all_grid_import(self) -> None:
"""Jeden slot buy<0 nesmí z min(buy) udělat všechny sloty expensive_import (gi=0 pro dům)."""
base = datetime(2026, 5, 22, 13, 15, tzinfo=timezone.utc)
slots = [
PlanningSlot(
interval_start=base + timedelta(minutes=15 * i),
buy_price=-0.54 if i == 15 else (0.8 + i * 0.05),
sell_price=-0.06 if i < 3 else 2.0,
pv_a_forecast_w=0,
pv_b_forecast_w=max(0, 5000 - i * 100) if i < 25 else 0,
load_baseline_w=5316 if i < 10 else 3392,
ev1_connected=False,
ev2_connected=False,
allow_charge=(i == 15 or i < 3),
allow_discharge_export=False,
charge_acquisition_buy_czk_kwh=0.94,
future_sell_opportunity_czk_kwh=5.5,
)
for i in range(20)
]
battery = _battery(uc_wh=64_000.0, min_pct=12.0, arb_pct=20.0)
battery.planner_terminal_soc_value_factor = 0.0
battery.planner_discharge_floor_percent = 5.0
hp = SimpleNamespace(rated_heating_power_w=0, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0)
grid = SimpleNamespace(max_import_power_w=17_000, max_export_power_w=13_500)
vehicles = [
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
]
results, _, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
float(battery.soc_max_wh),
50.0,
operating_mode="AUTO",
)
self.assertEqual(len(results), 20)
def test_spot_low_acquisition_does_not_mark_all_slots_expensive(self) -> None:
"""Spot + charge_acquisition ~0,9 nesmí z buy>acq udělat gi=0 pro dům ve všech slotech."""
base = datetime(2026, 5, 22, 10, 0, tzinfo=timezone.utc)
slots = [
PlanningSlot(
interval_start=base + timedelta(minutes=15 * i),
buy_price=2.5 + 0.1 * i,
sell_price=3.0,
pv_a_forecast_w=2000,
pv_b_forecast_w=3000,
load_baseline_w=2000,
ev1_connected=False,
ev2_connected=False,
allow_charge=False,
allow_discharge_export=False,
charge_acquisition_buy_czk_kwh=0.94,
future_sell_opportunity_czk_kwh=5.5,
)
for i in range(24)
]
battery = _battery(uc_wh=64_000.0, min_pct=12.0, arb_pct=20.0)
battery.planner_terminal_soc_value_factor = 0.0
hp = SimpleNamespace(rated_heating_power_w=0, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0)
grid = SimpleNamespace(max_import_power_w=17_000, max_export_power_w=13_500)
vehicles = [
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
]
results, _, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
30_000.0,
50.0,
operating_mode="AUTO",
)
self.assertEqual(len(results), 24)
class AutoPassiveNoLoadFollowingDischargeTests(unittest.TestCase):
"""AUTO bez allow_discharge_export: žádný export do sítě (Deye PASSIVE)."""
def test_no_grid_export_on_inflated_baseline_without_discharge_mask(self) -> None:
slots = [
PlanningSlot(
interval_start=datetime(2026, 5, 16, 9, 45, tzinfo=timezone.utc),
buy_price=0.77,
sell_price=0.09,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=8542,
ev1_connected=False,
ev2_connected=False,
is_predicted_price=False,
allow_charge=False,
allow_discharge_export=False,
)
]
battery = _battery(uc_wh=20_000.0, min_pct=10.0, arb_pct=20.0)
hp = SimpleNamespace(
rated_heating_power_w=0,
tuv_min_temp_c=45.0,
tuv_target_temp_c=55.0,
)
grid = SimpleNamespace(max_import_power_w=17_000, max_export_power_w=8000)
vehicles = [
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
]
soc0 = 0.45 * battery.usable_capacity_wh
results, _, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
soc0,
50.0,
tuv_delta_stats=None,
operating_mode="AUTO",
)
self.assertEqual(len(results), 1)
self.assertEqual(results[0].deye_physical_mode, "PASSIVE")
self.assertGreaterEqual(
results[0].grid_setpoint_w,
0,
msg="must not export to grid when allow_discharge_export=false",
)
class TerminalSocShadowTests(unittest.TestCase):
"""Terminal SoC shadow price v objective drží konec horizontu nad holým minimem."""
def test_terminal_soc_shadow_price_prevents_drain(self) -> None:
base = datetime(2026, 4, 3, 12, 0, tzinfo=timezone.utc)
slots = []
for i in range(3):
slots.append(
PlanningSlot(
interval_start=base + timedelta(minutes=15 * i),
buy_price=2.0,
sell_price=0.6,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=600,
ev1_connected=False,
ev2_connected=False,
is_predicted_price=False,
)
)
slots.append(
PlanningSlot(
interval_start=base + timedelta(minutes=45),
buy_price=2.0,
sell_price=14.0,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=600,
ev1_connected=False,
ev2_connected=False,
is_predicted_price=False,
)
)
battery = _battery(uc_wh=12_000.0, min_pct=12.0, arb_pct=20.0)
hp = SimpleNamespace(
rated_heating_power_w=0,
tuv_min_temp_c=45.0,
tuv_target_temp_c=55.0,
)
grid = SimpleNamespace(max_import_power_w=20_000, max_export_power_w=20_000)
vehicles = [
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
SimpleNamespace(
max_charge_power_w=0,
battery_capacity_kwh=1.0,
default_target_soc_pct=80.0,
),
]
soc0 = 0.5 * battery.usable_capacity_wh
results, _ms, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
soc0,
50.0,
tuv_delta_stats=None,
operating_mode="AUTO",
)
self.assertEqual(len(results), 4)
# Bez shadow price by solver mohl končit u min SoC; kotva drží znatelnou rezervu.
self.assertGreaterEqual(
results[-1].battery_soc_target,
15.0,
msg="terminal SoC shadow price should keep end-of-horizon SoC above bare minimum",
)
class SpreadGuardHome01EconomicsTests(unittest.TestCase):
"""Regrese: sell≪buy (VT) nesmí vést k PV exportu + masivnímu grid importu ve stejném slotu."""
def test_loss_making_morning_and_vt_slot_avoid_export_and_grid_charge(self) -> None:
from test_planning_charge_slot_selection import (
_battery as mask_battery,
_select_charge_slots,
_select_discharge_export_slots,
)
base = datetime(2026, 5, 21, 8, 0, tzinfo=timezone.utc)
raw: list[tuple[float, float, int, int]] = [
(1.55, 0.01, 6_000, 2_000),
(1.55, 0.01, 6_500, 2_000),
(1.49, -0.04, 0, 3_500),
(0.86, 0.01, 0, 3_500),
(0.86, 0.01, 0, 3_500),
(0.86, 0.01, 5_000, 2_000),
]
slots: list[PlanningSlot] = []
for i, (buy, sell, pv, load) in enumerate(raw):
slots.append(
PlanningSlot(
interval_start=base + timedelta(minutes=15 * i),
buy_price=buy,
sell_price=sell,
pv_a_forecast_w=0,
pv_b_forecast_w=pv,
load_baseline_w=load,
ev1_connected=False,
ev2_connected=False,
is_predicted_price=False,
)
)
mb = mask_battery(uc_wh=64_000.0)
soc0 = 0.31 * mb.usable_capacity_wh
charge = _select_charge_slots(slots, mb, soc0)
discharge = _select_discharge_export_slots(slots, mb, soc0)
for t, s in enumerate(slots):
s.allow_charge = t in charge
s.allow_discharge_export = t in discharge
battery = _battery(uc_wh=64_000.0, terminal_soc_value_factor=0.9)
hp = SimpleNamespace(rated_heating_power_w=0, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0)
grid = SimpleNamespace(max_import_power_w=20_000, max_export_power_w=20_000)
vehicles = [
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
]
results, _ms, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
soc0,
50.0,
operating_mode="AUTO",
)
self.assertEqual(len(results), len(slots))
morning = results[0]
vt_before_nt = results[2]
self.assertLessEqual(morning.grid_setpoint_w, slots[0].load_baseline_w + 4_500)
self.assertNotEqual(morning.export_mode, "PV_SURPLUS")
self.assertGreaterEqual(
vt_before_nt.grid_setpoint_w,
-6_500,
msg="před NT: žádný masivní export při téměř nulovém sell",
)
self.assertLessEqual(vt_before_nt.battery_setpoint_w, 10_500)
class ChargeAcquisitionArbitrageTests(unittest.TestCase):
"""Mezi-slotová arbitráž: večerní export při nízké charge_acquisition z SQL."""
def test_evening_battery_export_when_sell_above_acquisition(self) -> None:
base = datetime(2026, 5, 21, 10, 0, tzinfo=timezone.utc)
cheap = (0.75, 0.25)
peak = (7.0, 4.8)
slots: list[PlanningSlot] = []
for i in range(6):
buy, sell = cheap if i < 2 else peak
slots.append(
PlanningSlot(
interval_start=base + timedelta(minutes=15 * i),
buy_price=buy,
sell_price=sell,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=800,
ev1_connected=False,
ev2_connected=False,
is_predicted_price=False,
allow_charge=i < 2,
allow_discharge_export=i >= 2,
charge_acquisition_buy_czk_kwh=0.75,
charge_acquisition_cutoff_at=base + timedelta(minutes=30),
)
)
battery = _battery(uc_wh=64_000.0, terminal_soc_value_factor=0.0)
battery.max_charge_power_w = 17_000
battery.max_discharge_power_w = 17_000
hp = SimpleNamespace(rated_heating_power_w=0, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0)
grid = SimpleNamespace(max_import_power_w=20_000, max_export_power_w=20_000)
vehicles = [
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
]
soc0 = 0.78 * battery.usable_capacity_wh
results, _ms, snap = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
soc0,
50.0,
operating_mode="AUTO",
)
self.assertAlmostEqual(
snap["inputs"]["charge_acquisition_buy_czk_kwh"],
0.75,
places=2,
)
evening = results[3]
self.assertLess(
evening.grid_setpoint_w,
-1_000,
msg="high sell vs low acquisition should motivate grid export",
)
self.assertLess(evening.battery_setpoint_w, -500)
def test_no_pv_export_at_low_sell_when_evening_peak_much_higher(self) -> None:
"""Odpolední sell ~1,4 a večer ~5,5 — PV do baterie, ne FVE→síť za haléř."""
base = datetime(2026, 5, 21, 12, 0, tzinfo=timezone.utc)
afternoon = PlanningSlot(
interval_start=base,
buy_price=4.5,
sell_price=1.4,
pv_a_forecast_w=8000,
pv_b_forecast_w=0,
load_baseline_w=2500,
ev1_connected=False,
ev2_connected=False,
allow_charge=False,
allow_discharge_export=False,
charge_acquisition_buy_czk_kwh=0.82,
future_sell_opportunity_czk_kwh=5.5,
)
cheap = PlanningSlot(
interval_start=base + timedelta(hours=20),
buy_price=0.5,
sell_price=-0.2,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=2000,
ev1_connected=False,
ev2_connected=False,
allow_charge=True,
allow_discharge_export=False,
charge_acquisition_buy_czk_kwh=0.82,
future_sell_opportunity_czk_kwh=5.5,
)
peak = PlanningSlot(
interval_start=base + timedelta(hours=7),
buy_price=7.0,
sell_price=5.5,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=2500,
ev1_connected=False,
ev2_connected=False,
allow_charge=False,
allow_discharge_export=True,
charge_acquisition_buy_czk_kwh=0.82,
future_sell_opportunity_czk_kwh=5.5,
)
slots = [afternoon, peak, cheap]
battery = _battery(uc_wh=64_000.0, min_pct=12.0, arb_pct=20.0)
battery.max_charge_power_w = 18_000
battery.max_discharge_power_w = 18_000
hp = SimpleNamespace(rated_heating_power_w=0, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0)
grid = SimpleNamespace(max_import_power_w=17_000, max_export_power_w=13_500)
vehicles = [
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
]
soc0 = 0.5 * battery.usable_capacity_wh
results, _ms, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
soc0,
50.0,
operating_mode="AUTO",
)
pm = results[0]
self.assertGreaterEqual(
pm.grid_setpoint_w,
-50,
"low sell with high evening peak: keep PV for battery, not grid dump",
)
self.assertGreater(
pm.battery_setpoint_w,
500,
"PV surplus should charge battery ahead of evening export",
)
class Home01RegressionTests(unittest.TestCase):
"""Definition of Done: home-01 arbitráž archetypy (bez DB)."""
@staticmethod
def _solve_auto(
slots: list[PlanningSlot],
battery: SimpleNamespace,
soc0: float,
*,
two_pass: bool = True,
) -> tuple[list[DispatchResult], dict]:
hp = SimpleNamespace(rated_heating_power_w=0, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0)
grid = SimpleNamespace(max_import_power_w=20_000, max_export_power_w=20_000)
vehicles = [
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
]
fn = solve_dispatch_two_pass if two_pass else solve_dispatch
results, _ms, snap = fn(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
soc0,
50.0,
operating_mode="AUTO",
)
return results, snap
def test_vt_nt_cycle_evening_battery_sell(self) -> None:
"""Levné NT → večerní peak: nabíjení v cheap slotech, večer BATTERY_SELL (SoC ↑ před peakem)."""
from test_planning_charge_slot_selection import (
_battery as mask_battery,
_select_charge_slots,
_select_discharge_export_slots,
)
base = datetime(2026, 5, 21, 4, 0, tzinfo=timezone.utc)
prices: list[tuple[float, float, int, int]] = [
(0.42, -0.20, 0, 2300),
(0.44, -0.19, 0, 2350),
(0.46, -0.18, 0, 2380),
(0.48, -0.18, 0, 2400),
(0.50, -0.15, 0, 2600),
(0.52, -0.14, 0, 2700),
(0.55, -0.12, 0, 2800),
(0.58, -0.11, 0, 2850),
(0.62, -0.10, 0, 2900),
(0.68, -0.09, 0, 2950),
(0.72, -0.08, 500, 3000),
(0.76, -0.07, 1500, 3100),
(0.80, -0.05, 2000, 3200),
(7.20, 5.50, 0, 2500),
(7.00, 5.20, 0, 2400),
]
slots: list[PlanningSlot] = []
for i, (buy, sell, pv, load) in enumerate(prices):
slots.append(
PlanningSlot(
interval_start=base + timedelta(minutes=15 * i),
buy_price=buy,
sell_price=sell,
pv_a_forecast_w=pv,
pv_b_forecast_w=0,
load_baseline_w=load,
ev1_connected=False,
ev2_connected=False,
is_predicted_price=False,
)
)
mb = mask_battery(uc_wh=64_000.0, charge_buf=1.5, discharge_buf=1.0)
soc0 = 0.10 * mb.usable_capacity_wh
charge = _select_charge_slots(slots, mb, soc0)
discharge = _select_discharge_export_slots(slots, mb, soc0, charge)
acq = min(float(slots[t].buy_price) for t in charge) if charge else 0.9
cutoff = min(
(slots[t].interval_start for t in discharge),
default=slots[-1].interval_start,
)
for t, s in enumerate(slots):
s.allow_charge = t in charge or float(s.buy_price) < 1.0
# Export jen při skutečné večerní špičce (sell ≥ 5), ne při mezilehlém 4.8 Kč.
s.allow_discharge_export = t in discharge and float(s.sell_price) >= 5.0
s.charge_acquisition_buy_czk_kwh = acq
s.charge_acquisition_cutoff_at = cutoff
battery = _battery(uc_wh=64_000.0, min_pct=12.0, arb_pct=20.0, terminal_soc_value_factor=0.2)
battery.max_charge_power_w = 17_000
battery.max_discharge_power_w = 17_000
soc_start_pct = 100.0 * soc0 / battery.usable_capacity_wh
results, snap = self._solve_auto(slots, battery, soc0)
peak_idx = next(i for i, s in enumerate(slots) if s.sell_price >= 5.0)
pre_peak = results[peak_idx - 1] if peak_idx > 0 else results[0]
self.assertGreater(
pre_peak.battery_soc_target,
soc_start_pct + 25.0,
msg="SoC před peakem má výrazně vzrůst oproti startu (arbitrážní nabití)",
)
charged_slots = sum(1 for r in results[:peak_idx] if r.battery_setpoint_w > 500 or r.grid_setpoint_w > 500)
self.assertGreater(charged_slots, 2, "levné sloty mají nabíjet ze sítě nebo PV")
evening = results[peak_idx]
total_export_w = max(0, -evening.grid_setpoint_w) + max(0, -evening.battery_setpoint_w)
self.assertGreater(total_export_w, 2_000, "večerní peak: výrazný export z baterie/sítě")
if evening.grid_setpoint_w < 0:
self.assertEqual(evening.export_mode, "BATTERY_SELL")
inputs = snap.get("inputs") or {}
self.assertTrue(inputs.get("two_pass_enabled"))
def test_neg_sell_pv_to_battery_not_grid_when_soc_has_room(self) -> None:
"""sell<0, spot, PV B: při SoC pod stropem jen nabíjení/curtail, ne PV_SURPLUS export."""
slots = [
PlanningSlot(
interval_start=datetime(2026, 5, 25, 8, 0, tzinfo=timezone.utc)
+ timedelta(minutes=15 * i),
buy_price=0.5,
sell_price=-0.4,
pv_a_forecast_w=8000,
pv_b_forecast_w=2000,
load_baseline_w=400,
ev1_connected=False,
ev2_connected=False,
allow_charge=True,
allow_discharge_export=False,
)
for i in range(4)
]
battery = _battery(uc_wh=64_000.0, terminal_soc_value_factor=0.2)
battery.max_charge_power_w = 18_000
grid = SimpleNamespace(
max_import_power_w=17_000,
max_export_power_w=13_500,
block_export_on_negative_sell=False,
purchase_pricing_mode="spot",
)
hp = SimpleNamespace(rated_heating_power_w=0, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0)
vehicles = [
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
]
results, _ms, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
30_000.0,
50.0,
operating_mode="AUTO",
)
for r in results:
self.assertGreaterEqual(r.grid_setpoint_w, 0, "neg sell bez exportu při volné kapacitě baterie")
self.assertGreater(r.battery_setpoint_w, 0, "neg sell má nabíjet z FVE")
def test_neg_sell_full_battery_exports_at_most_pv_b_not_full_surplus(self) -> None:
"""Plná baterie + sell<0: max export jen pole B (~5 kW), ne pv_a+pv_b (~9 kW)."""
slots = [
PlanningSlot(
interval_start=datetime(2026, 5, 25, 7, 30, tzinfo=timezone.utc)
+ timedelta(minutes=15 * i),
buy_price=0.5,
sell_price=-0.4,
pv_a_forecast_w=4700,
pv_b_forecast_w=5100,
load_baseline_w=400,
ev1_connected=False,
ev2_connected=False,
allow_charge=True,
allow_discharge_export=False,
)
for i in range(3)
]
battery = _battery(uc_wh=64_000.0, terminal_soc_value_factor=0.2)
battery.max_charge_power_w = 18_000
battery.soc_max_wh = 64_000.0
grid = SimpleNamespace(
max_import_power_w=17_000,
max_export_power_w=13_500,
block_export_on_negative_sell=False,
purchase_pricing_mode="spot",
)
hp = SimpleNamespace(rated_heating_power_w=0, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0)
vehicles = [
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
]
soc0 = float(battery.soc_max_wh) - 500.0
results, _ms, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
soc0,
50.0,
operating_mode="AUTO",
)
for r in results:
export_w = max(0, -int(r.grid_setpoint_w or 0))
if export_w > 0:
self.assertLessEqual(
export_w,
5_500,
"při plné baterii jen ventil pole B, ne celý PV přebytek",
)
def test_neg_sell_bat_dump_slot_selection(self) -> None:
"""sell<0 těsně před buy<=-2: slot je v neg_sell_bat_dump_slots (ge_bat povolen)."""
from services.planning_engine import _neg_sell_bat_dump_slots
slots = [
PlanningSlot(
interval_start=datetime(2026, 4, 4, 5, 0, tzinfo=timezone.utc),
buy_price=0.3,
sell_price=-0.35,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=800,
ev1_connected=False,
ev2_connected=False,
allow_charge=False,
allow_discharge_export=False,
),
PlanningSlot(
interval_start=datetime(2026, 4, 4, 5, 15, tzinfo=timezone.utc),
buy_price=-10.0,
sell_price=-0.2,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=800,
ev1_connected=False,
ev2_connected=False,
allow_charge=True,
allow_discharge_export=False,
),
]
grid = SimpleNamespace(
block_export_on_negative_sell=False,
purchase_pricing_mode="spot",
)
dump = _neg_sell_bat_dump_slots(
slots,
operating_mode="AUTO",
purchase_fixed=False,
grid=grid,
buy_extreme_thr=-2.0,
degrad_czk_kwh=0.15,
)
self.assertEqual(dump, {0})
def test_no_fve_dump_at_low_sell_with_evening_peak(self) -> None:
"""Odpolední sell ~1,4 vs večer ~5,5 — žádný PV_SURPLUS export, nabíjení z FVE."""
base = datetime(2026, 5, 21, 14, 0, tzinfo=timezone.utc)
afternoon = PlanningSlot(
interval_start=base,
buy_price=4.5,
sell_price=1.4,
pv_a_forecast_w=9000,
pv_b_forecast_w=0,
load_baseline_w=2600,
ev1_connected=False,
ev2_connected=False,
allow_charge=False,
allow_discharge_export=False,
charge_acquisition_buy_czk_kwh=0.78,
future_sell_opportunity_czk_kwh=5.5,
)
peak = PlanningSlot(
interval_start=base + timedelta(hours=5),
buy_price=7.0,
sell_price=5.5,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=2400,
ev1_connected=False,
ev2_connected=False,
allow_charge=False,
allow_discharge_export=True,
charge_acquisition_buy_czk_kwh=0.78,
future_sell_opportunity_czk_kwh=5.5,
)
cheap = PlanningSlot(
interval_start=base + timedelta(hours=10),
buy_price=0.55,
sell_price=-0.1,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=2000,
ev1_connected=False,
ev2_connected=False,
allow_charge=True,
allow_discharge_export=False,
charge_acquisition_buy_czk_kwh=0.78,
future_sell_opportunity_czk_kwh=5.5,
)
slots = [afternoon, peak, cheap]
battery = _battery(uc_wh=64_000.0)
battery.max_charge_power_w = 18_000
soc0 = 0.48 * battery.usable_capacity_wh
results, _ = self._solve_auto(slots, battery, soc0)
pm = results[0]
self.assertNotEqual(pm.export_mode, "PV_SURPLUS")
self.assertGreater(pm.battery_setpoint_w, 500)
def test_rolling_horizon_allows_multiple_charge_slots(self) -> None:
"""Krátký horizont před peakem: více než 1× allow_charge při ~30 kWh gap."""
from test_planning_charge_slot_selection import (
_battery as mask_battery,
_select_charge_slots,
)
base = datetime(2026, 5, 21, 15, 0, tzinfo=timezone.utc)
slots: list[PlanningSlot] = []
for i in range(5):
buy = 0.65 + 0.05 * i if i < 3 else 6.0
sell = -0.1 if i < 3 else 5.2
slots.append(
PlanningSlot(
interval_start=base + timedelta(minutes=15 * i),
buy_price=buy,
sell_price=sell,
pv_a_forecast_w=1500,
pv_b_forecast_w=0,
load_baseline_w=3000,
ev1_connected=False,
ev2_connected=False,
is_predicted_price=False,
)
)
mb = mask_battery(uc_wh=64_000.0, charge_buf=1.3)
soc0 = 0.22 * mb.usable_capacity_wh
charge = _select_charge_slots(slots, mb, soc0)
self.assertGreaterEqual(
len(charge),
2,
msg="při velkém energy_to_fill má maska vybrat více levných slotů",
)
def test_negative_sell_blocks_export(self) -> None:
base = datetime(2026, 5, 21, 10, 0, tzinfo=timezone.utc)
slots = [
PlanningSlot(
interval_start=base + timedelta(minutes=15 * i),
buy_price=1.0,
sell_price=-0.8 if i < 2 else 2.0,
pv_a_forecast_w=5000,
pv_b_forecast_w=0,
load_baseline_w=2000,
ev1_connected=False,
ev2_connected=False,
is_predicted_price=False,
)
for i in range(4)
]
battery = _battery(uc_wh=40_000.0)
results, _ = self._solve_auto(slots, battery, 0.5 * battery.usable_capacity_wh)
for i in range(2):
self.assertGreaterEqual(results[i].grid_setpoint_w, -50)
self.assertNotEqual(results[i].export_mode, "PV_SURPLUS")
@staticmethod
def _home01_run16522_slots() -> list[PlanningSlot]:
from test_planning_charge_slot_selection import (
_battery as mask_battery,
_select_charge_slots,
_select_discharge_export_slots,
)
from zoneinfo import ZoneInfo
prague = ZoneInfo("Europe/Prague")
base = datetime(2026, 5, 24, 0, 0, tzinfo=prague)
hour_specs: list[tuple[int, int, dict]] = [
(0, 5, {"buy": 4.7, "sell": 2.9}),
(5, 7, {"buy": 5.0, "sell": 3.0, "pv_b": 400}),
(7, 11, {"buy": 4.5, "sell": 2.8, "pv_a": 3000, "pv_b": 2000}),
(11, 14, {"buy": 0.5, "sell": -0.4, "pv_a": 6000, "pv_b": 5000}),
(14, 17, {"buy": 1.0, "sell": -0.3, "pv_a": 5000, "pv_b": 4000}),
(17, 19, {"buy": 4.5, "sell": 3.0}),
(19, 22, {"buy": 6.5, "sell": 4.0}),
(22, 24, {"buy": 4.8, "sell": 3.0}),
]
slots: list[PlanningSlot] = []
for h0, h1, kw in hour_specs:
for h in range(h0, h1):
for minute in (0, 15, 30, 45):
t = base.replace(hour=h, minute=minute).astimezone(timezone.utc)
slots.append(
PlanningSlot(
interval_start=t,
buy_price=float(kw["buy"]),
sell_price=float(kw["sell"]),
pv_a_forecast_w=int(kw.get("pv_a", 0)),
pv_b_forecast_w=int(kw.get("pv_b", 0)),
load_baseline_w=500,
ev1_connected=False,
ev2_connected=False,
is_predicted_price=False,
)
)
mb = mask_battery(charge_buf=1.3, uc_wh=64_000.0, soc_max_pct=95.0)
soc0 = 30_000.0
charge = _select_charge_slots(slots, mb, soc0)
discharge = _select_discharge_export_slots(slots, mb, soc0, charge)
acq = (
sum(float(slots[t].buy_price) for t in charge) / len(charge)
if charge
else min(float(s.buy_price) for s in slots)
)
cutoff = min(
(slots[t].interval_start for t in discharge),
default=slots[-1].interval_start,
)
for t, s in enumerate(slots):
s.allow_charge = t in charge or float(s.buy_price) < 0
s.allow_discharge_export = t in discharge
s.charge_acquisition_buy_czk_kwh = acq
s.charge_acquisition_cutoff_at = cutoff
return slots
def _home01_battery(self, soc: float = 30_000.0) -> SimpleNamespace:
b = _battery(
uc_wh=64_000.0,
min_pct=11.0,
arb_pct=20.0,
terminal_soc_value_factor=0.2,
)
b.max_charge_power_w = 17_000
b.max_discharge_power_w = 17_000
b.charge_slot_buffer = 1.3
b.planner_daytime_charge_target_enabled = True
return b
def _home01_grid(self) -> SimpleNamespace:
return SimpleNamespace(
max_import_power_w=17_000,
max_export_power_w=13_500,
block_export_on_negative_sell=False,
purchase_pricing_mode="spot",
)
def test_home01_no_night_charge_before_pv_day(self) -> None:
"""Pattern run 16522: 22:00-24:00 bez grid importu >15 kW pred PV dnem."""
from zoneinfo import ZoneInfo
slots = self._home01_run16522_slots()
results, _snap = self._solve_auto(
slots,
self._home01_battery(),
30_000.0,
)
prague = ZoneInfo("Europe/Prague")
for r in results:
h = r.interval_start.astimezone(prague).hour
if h in (22, 23):
self.assertLess(
r.grid_setpoint_w,
15_000,
f"slot {r.interval_start}: grid={r.grid_setpoint_w} >= 15 kW",
)
def test_two_pass_converged_after_filter(self) -> None:
"""Po self-konzistentni masce B: acquisition pass1 ~ pass2."""
slots = self._home01_run16522_slots()
_results, snap = self._solve_auto(slots, self._home01_battery(), 30_000.0)
inputs = snap.get("inputs") or {}
self.assertTrue(
inputs.get("two_pass_converged"),
f"acquisition diverguje: {inputs}",
)
class LoadFirstDispatchTests(unittest.TestCase):
"""Deye load-first: PV do spotřeby dřív než bc_pv/ge_pv z přebytku."""
@staticmethod
def _solve_auto(
slots: list[PlanningSlot],
battery: SimpleNamespace,
soc0: float,
) -> list[DispatchResult]:
hp = SimpleNamespace(rated_heating_power_w=0, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0)
grid = SimpleNamespace(max_import_power_w=20_000, max_export_power_w=20_000)
vehicles = [
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
]
results, _, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
soc0,
50.0,
operating_mode="AUTO",
)
return results
def test_high_pv_low_load_prefers_export_over_battery_charge(self) -> None:
"""Mimo grid-charge masku nesmí LP nabíjet z celého PV při malé zátěži."""
base = datetime(2026, 5, 21, 12, 0, tzinfo=timezone.utc)
slots = [
PlanningSlot(
interval_start=base,
buy_price=2.0,
sell_price=4.0,
pv_a_forecast_w=8000,
pv_b_forecast_w=0,
load_baseline_w=500,
ev1_connected=False,
ev2_connected=False,
is_predicted_price=False,
allow_charge=False,
allow_discharge_export=False,
)
]
battery = _battery(uc_wh=50_000.0)
soc0 = 0.5 * battery.usable_capacity_wh
r = self._solve_auto(slots, battery, soc0)[0]
self.assertLessEqual(
r.battery_setpoint_w,
200,
msg="load-first: přebytek FVE má jít do exportu, ne do bc_pv",
)
self.assertLess(
r.grid_setpoint_w,
-400,
msg="očekáván PV export (přebytek po load-first)",
)
self.assertEqual(r.export_mode, "PV_SURPLUS")
class PreNegativeSellExportTests(unittest.TestCase):
"""Před prvním sell<0: export přebytku (BA81/KV1 strategie), ne nabíjení + pozdní vývoz."""
def test_kv1_like_morning_exports_before_negative_sell_window(self) -> None:
base = datetime(2026, 5, 22, 6, 45, tzinfo=timezone.utc)
slots = [
PlanningSlot(
interval_start=base + timedelta(minutes=15 * i),
buy_price=6.35,
sell_price=2.2,
pv_a_forecast_w=5000,
pv_b_forecast_w=0,
load_baseline_w=400,
ev1_connected=False,
ev2_connected=False,
allow_charge=False,
allow_discharge_export=False,
charge_acquisition_buy_czk_kwh=6.35,
future_sell_opportunity_czk_kwh=5.5,
)
for i in range(8)
] + [
PlanningSlot(
interval_start=base + timedelta(hours=2),
buy_price=6.35,
sell_price=-0.3,
pv_a_forecast_w=6000,
pv_b_forecast_w=0,
load_baseline_w=400,
ev1_connected=False,
ev2_connected=False,
allow_charge=False,
allow_discharge_export=False,
charge_acquisition_buy_czk_kwh=6.35,
future_sell_opportunity_czk_kwh=-0.3,
),
]
battery = _battery(uc_wh=12_500.0, min_pct=10.0, arb_pct=30.0)
battery.max_charge_power_w = 6250
hp = SimpleNamespace(rated_heating_power_w=0, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0)
grid = SimpleNamespace(
max_import_power_w=17_000,
max_export_power_w=8000,
block_export_on_negative_sell=True,
)
vehicles = [
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
]
soc0 = 0.85 * battery.soc_max_wh
results, _, _ = solve_dispatch(
slots, battery, hp, grid, [None, None], vehicles, soc0, 50.0, operating_mode="AUTO"
)
self.assertLess(results[0].grid_setpoint_w, -500, "ráno: přebytek FVE do sítě před sell<0")
neg = results[8]
self.assertGreater(neg.battery_setpoint_w, 500, "záporný sell: PV do baterie")
self.assertEqual(neg.export_mode, "NONE")
class Home01PvStoreValueTests(unittest.TestCase):
"""FVE (zejména pole B) nesmí jít do sítě pod hodnotou uložení / večerní peak."""
def test_pv_b_low_sell_charges_not_exports(self) -> None:
"""08:30 archetyp: sell ~0,09, večer ~5,5 → bc, ne ge_pv."""
slots = [
PlanningSlot(
interval_start=datetime(2026, 5, 22, 6, 30, tzinfo=timezone.utc),
buy_price=1.017,
sell_price=0.088,
pv_a_forecast_w=0,
pv_b_forecast_w=5313,
load_baseline_w=1961,
ev1_connected=False,
ev2_connected=False,
allow_charge=True,
allow_discharge_export=False,
charge_acquisition_buy_czk_kwh=0.526,
future_sell_opportunity_czk_kwh=5.5,
)
]
battery = _battery(uc_wh=64_000.0, min_pct=12.0, arb_pct=20.0)
battery.max_charge_power_w = 18_000
hp = SimpleNamespace(rated_heating_power_w=0, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0)
grid = SimpleNamespace(max_import_power_w=17_000, max_export_power_w=13_500)
vehicles = [
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
]
soc0 = 0.45 * battery.usable_capacity_wh
results, _, _ = solve_dispatch(
slots, battery, hp, grid, [None, None], vehicles, soc0, 50.0, operating_mode="AUTO"
)
r = results[0]
self.assertGreaterEqual(r.grid_setpoint_w, 0, "nízký sell: žádný export FVE")
self.assertGreater(r.battery_setpoint_w, 500, "přebytek PV do baterie")
def test_negative_sell_no_pv_export_when_battery_has_room(self) -> None:
slots = [
PlanningSlot(
interval_start=datetime(2026, 5, 22, 7, 45, tzinfo=timezone.utc),
buy_price=0.55,
sell_price=-0.266,
pv_a_forecast_w=0,
pv_b_forecast_w=5474,
load_baseline_w=1961,
ev1_connected=False,
ev2_connected=False,
allow_charge=True,
allow_discharge_export=False,
charge_acquisition_buy_czk_kwh=0.526,
future_sell_opportunity_czk_kwh=5.5,
)
]
battery = _battery(uc_wh=64_000.0)
battery.max_charge_power_w = 18_000
hp = SimpleNamespace(rated_heating_power_w=0, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0)
grid = SimpleNamespace(max_import_power_w=17_000, max_export_power_w=13_500)
vehicles = [
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
]
soc0 = 0.5 * battery.usable_capacity_wh
results, _, _ = solve_dispatch(
slots, battery, hp, grid, [None, None], vehicles, soc0, 50.0, operating_mode="AUTO"
)
self.assertGreaterEqual(results[0].grid_setpoint_w, 0)
class SitePowerCapTests(unittest.TestCase):
"""Tvrdé limity site import a součtu nabíjení baterie."""
def test_grid_charge_respects_import_and_battery_caps(self) -> None:
"""home-01 typ: CHARGE slot nesmí překročit 17 kW import ani 18 kW do baterie."""
base = datetime(2026, 5, 22, 8, 45, tzinfo=timezone.utc)
slots = [
PlanningSlot(
interval_start=base,
buy_price=0.7,
sell_price=2.5,
pv_a_forecast_w=5000,
pv_b_forecast_w=0,
load_baseline_w=1961,
ev1_connected=False,
ev2_connected=False,
is_predicted_price=False,
allow_charge=True,
allow_discharge_export=False,
)
]
battery = _battery(uc_wh=64_000.0)
battery.max_charge_power_w = 18_000
battery.max_discharge_power_w = 18_000
hp = SimpleNamespace(rated_heating_power_w=0, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0)
grid = SimpleNamespace(max_import_power_w=17_000, max_export_power_w=13_500)
vehicles = [
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
]
soc0 = 0.5 * battery.usable_capacity_wh
results, _, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
soc0,
50.0,
operating_mode="AUTO",
)
r = results[0]
self.assertLessEqual(
r.grid_setpoint_w,
17_000,
msg="import ze site ≤ max_import_power_w",
)
self.assertGreaterEqual(r.grid_setpoint_w, 0)
self.assertLessEqual(
r.battery_setpoint_w,
18_000,
msg="nabíjení baterie ≤ max_charge_power_w",
)
self.assertGreater(r.battery_setpoint_w, 0)
def test_battery_export_respects_site_export_cap(self) -> None:
"""SELL slot: vývoz ze site ≤ max_export; vybíjení baterie ≤ max_discharge."""
base = datetime(2026, 5, 22, 18, 0, tzinfo=timezone.utc)
slots = [
PlanningSlot(
interval_start=base,
buy_price=0.5,
sell_price=6.0,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=2500,
ev1_connected=False,
ev2_connected=False,
is_predicted_price=False,
allow_charge=False,
allow_discharge_export=True,
)
]
battery = _battery(uc_wh=64_000.0, min_pct=12.0, arb_pct=20.0)
battery.max_charge_power_w = 18_000
battery.max_discharge_power_w = 18_000
hp = SimpleNamespace(rated_heating_power_w=0, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0)
grid = SimpleNamespace(max_import_power_w=17_000, max_export_power_w=13_500)
vehicles = [
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
]
soc0 = 0.85 * battery.usable_capacity_wh
results, _, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
soc0,
50.0,
operating_mode="AUTO",
)
r = results[0]
self.assertLessEqual(
-r.grid_setpoint_w,
13_500,
msg="export ze site ≤ max_export_power_w",
)
self.assertLess(r.grid_setpoint_w, -500, msg="očekáván významný export")
self.assertLess(r.battery_setpoint_w, -500, msg="očekáváno vybíjení baterie")
self.assertLessEqual(
r.export_limit_w,
13_500,
msg="export_limit_w odpovídá site limitu",
)
class PlannerArbitrageImprovementsTests(unittest.TestCase):
"""Regrese: záporný buy, peak sell před sell<0, večerní export cap."""
def test_pre_neg_peak_ignores_earlier_day_in_horizon(self) -> None:
"""Horizont přes půlnoc: peak je na dni záporného sell, ne včerejší večer."""
base = datetime(2026, 5, 22, 18, 0, tzinfo=timezone.utc)
slots = [
PlanningSlot(
interval_start=base,
buy_price=4.0,
sell_price=4.6,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=1000,
ev1_connected=False,
ev2_connected=False,
),
PlanningSlot(
interval_start=base + timedelta(hours=12),
buy_price=4.0,
sell_price=3.06,
pv_a_forecast_w=3000,
pv_b_forecast_w=0,
load_baseline_w=1000,
ev1_connected=False,
ev2_connected=False,
),
PlanningSlot(
interval_start=base + timedelta(hours=12, minutes=45),
buy_price=0.5,
sell_price=-0.1,
pv_a_forecast_w=5000,
pv_b_forecast_w=0,
load_baseline_w=1000,
ev1_connected=False,
ev2_connected=False,
),
]
first_neg = 2
self.assertEqual(_pre_neg_peak_sell_idx(slots, first_neg), 1)
def test_pre_neg_peak_ignores_midnight_on_same_day(self) -> None:
"""Půlnoc může mít vyšší sell než ráno — peak musí být v pásmu 511, ne 00:00."""
base = datetime(2026, 5, 22, 22, 0, tzinfo=timezone.utc)
slots = [
PlanningSlot(
interval_start=base + timedelta(minutes=15 * i),
buy_price=4.0,
sell_price=3.72 if i == 0 else (3.06 if i == 28 else 2.0),
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=1000,
ev1_connected=False,
ev2_connected=False,
)
for i in range(36)
] + [
PlanningSlot(
interval_start=base + timedelta(minutes=15 * 36),
buy_price=0.5,
sell_price=-0.1,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=1000,
ev1_connected=False,
ev2_connected=False,
),
]
first_neg = 36
peak_idx = _pre_neg_peak_sell_idx(slots, first_neg)
self.assertIsNotNone(peak_idx)
self.assertGreater(_prague_hour(slots[peak_idx]), 4)
self.assertLess(_prague_hour(slots[peak_idx]), 12)
def test_pre_neg_peak_idx_is_highest_positive_sell(self) -> None:
base = datetime(2026, 5, 23, 4, 0, tzinfo=timezone.utc)
slots = [
PlanningSlot(
interval_start=base + timedelta(minutes=15 * i),
buy_price=4.0,
sell_price=3.06 if i == 1 else (1.99 if i == 3 else 2.5),
pv_a_forecast_w=1000,
pv_b_forecast_w=0,
load_baseline_w=1000,
ev1_connected=False,
ev2_connected=False,
)
for i in range(6)
] + [
PlanningSlot(
interval_start=base + timedelta(minutes=15 * 6),
buy_price=0.5,
sell_price=-0.1,
pv_a_forecast_w=4000,
pv_b_forecast_w=0,
load_baseline_w=1000,
ev1_connected=False,
ev2_connected=False,
),
]
self.assertEqual(_pre_neg_peak_sell_idx(slots, 6), 1)
def test_morning_battery_export_at_peak_sell_before_negative_window(self) -> None:
base = datetime(2026, 5, 23, 4, 0, tzinfo=timezone.utc)
sells = [2.5, 3.06, 2.8, 1.99, 1.3, 0.34]
slots = [
PlanningSlot(
interval_start=base + timedelta(minutes=15 * i),
buy_price=4.0,
sell_price=sell,
pv_a_forecast_w=3000,
pv_b_forecast_w=0,
load_baseline_w=1000,
ev1_connected=False,
ev2_connected=False,
allow_charge=False,
allow_discharge_export=(i == 1),
future_sell_opportunity_czk_kwh=3.06,
)
for i, sell in enumerate(sells)
] + [
PlanningSlot(
interval_start=base + timedelta(minutes=15 * len(sells)),
buy_price=0.5,
sell_price=-0.1,
pv_a_forecast_w=5000,
pv_b_forecast_w=0,
load_baseline_w=1000,
ev1_connected=False,
ev2_connected=False,
allow_charge=True,
allow_discharge_export=False,
)
]
battery = _battery(uc_wh=64_000.0, min_pct=10.0, arb_pct=20.0)
battery.planner_discharge_floor_percent = 5.0
battery.max_discharge_power_w = 18_000
hp = SimpleNamespace(rated_heating_power_w=0, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0)
grid = SimpleNamespace(max_import_power_w=17_000, max_export_power_w=13_500)
vehicles = [
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
]
results, _, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
0.5 * battery.soc_max_wh,
50.0,
operating_mode="AUTO",
)
peak_export = max(0, -results[1].grid_setpoint_w) + max(0, -results[1].battery_setpoint_w)
late_export = max(0, -results[3].grid_setpoint_w) + max(0, -results[3].battery_setpoint_w)
self.assertGreater(peak_export, late_export)
def test_negative_buy_grid_charge_without_allow_charge_mask(self) -> None:
base = datetime(2026, 5, 23, 11, 0, tzinfo=timezone.utc)
slots = [
PlanningSlot(
interval_start=base,
buy_price=-0.54,
sell_price=-1.25,
pv_a_forecast_w=8000,
pv_b_forecast_w=5000,
load_baseline_w=2000,
ev1_connected=False,
ev2_connected=False,
allow_charge=False,
allow_discharge_export=False,
)
]
battery = _battery(uc_wh=64_000.0)
battery.max_charge_power_w = 18_000
hp = SimpleNamespace(rated_heating_power_w=0, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0)
grid = SimpleNamespace(max_import_power_w=17_000, max_export_power_w=13_500)
vehicles = [
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
]
results, _, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
0.4 * battery.soc_max_wh,
50.0,
operating_mode="AUTO",
)
r = results[0]
self.assertGreater(r.grid_setpoint_w, 3000)
self.assertGreater(r.battery_setpoint_w, 1000)
def test_high_sell_discharge_slot_pushes_export_toward_site_cap(self) -> None:
base = datetime(2026, 5, 23, 18, 0, tzinfo=timezone.utc)
slots = [
PlanningSlot(
interval_start=base + timedelta(minutes=15 * i),
buy_price=5.0,
sell_price=4.6,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=1500,
ev1_connected=False,
ev2_connected=False,
allow_charge=False,
allow_discharge_export=True,
charge_acquisition_buy_czk_kwh=0.8,
future_sell_opportunity_czk_kwh=2.0,
)
for i in range(3)
]
battery = _battery(uc_wh=64_000.0, min_pct=12.0, arb_pct=20.0)
battery.planner_terminal_soc_value_factor = 0.15
battery.max_discharge_power_w = 18_000
hp = SimpleNamespace(rated_heating_power_w=0, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0)
grid = SimpleNamespace(max_import_power_w=17_000, max_export_power_w=13_500)
vehicles = [
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0),
]
results, _, _ = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
0.8 * battery.soc_max_wh,
50.0,
operating_mode="AUTO",
)
r = results[1]
total_export = max(0, -r.grid_setpoint_w) + max(0, -r.battery_setpoint_w)
self.assertGreaterEqual(total_export, 11_000)
self.assertEqual(r.export_mode, "BATTERY_SELL")
if __name__ == "__main__":
unittest.main()