Files
ems/backend/tests/test_planning_dispatch_milp.py
Dusan Vojacek 8494ea26de
Some checks failed
CI and deploy / migration-check (push) Failing after 28s
CI and deploy / deploy (push) Has been skipped
nerezta PV A pri prodeji z baterie
2026-05-26 07:34:52 +02:00

3675 lines
138 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 zoneinfo import ZoneInfo
from services.planning_engine import (
DispatchResult,
PlanningSlot,
_dynamic_arb_floor_wh_series,
_dispatch_result_comparison,
_evening_battery_export_push_indices,
_evening_peak_export_indices,
_evening_push_discharge_budget_wh,
_in_night_battery_export_window,
_pre_neg_buy_soc_ceiling_wh,
_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,
discharge_slot_buffer: float = 1.5,
) -> 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,
discharge_slot_buffer=discharge_slot_buffer,
planner_terminal_soc_value_factor=terminal_soc_value_factor,
)
class PreNegBuySocPhaseTests(unittest.TestCase):
"""Dvoufázová SoC: plná při posledním sell≥0 před buy<0, strop před buy<0."""
def test_soc_ceiling_accounts_for_neg_buy_window(self) -> None:
base = datetime(2026, 5, 25, 8, 0, tzinfo=timezone.utc)
slots: list[PlanningSlot] = []
for i in range(16):
buy = -0.1 if 6 <= i < 10 else 1.0
sell = -0.3 if i < 6 else (2.5 if i < 10 else -0.2)
slots.append(
PlanningSlot(
interval_start=base + timedelta(minutes=15 * i),
buy_price=buy,
sell_price=sell,
pv_a_forecast_w=6000 if i >= 6 else 4000,
pv_b_forecast_w=3000 if i >= 6 else 2000,
load_baseline_w=2000,
ev1_connected=False,
ev2_connected=False,
allow_charge=True,
allow_discharge_export=True,
)
)
bat = _battery(uc_wh=64_000.0, min_pct=10.0, max_pct=95.0)
ceiling = _pre_neg_buy_soc_ceiling_wh(
slots,
first_neg_buy_idx=6,
min_soc_wh=bat.min_soc_wh,
soc_max_wh=bat.soc_max_wh,
max_charge_w=18_000,
charge_eff=0.95,
)
self.assertIsNotNone(ceiling)
assert ceiling is not None
self.assertLess(ceiling, bat.soc_max_wh * 0.85)
class EveningPushBudgetTests(unittest.TestCase):
"""Večerní tvrdý push: počet slotů z rozpočtu Wh (ne pevné top-3)."""
@staticmethod
def _evening_slots(n: int = 8) -> list[PlanningSlot]:
base = datetime(2026, 5, 25, 15, 0, tzinfo=timezone.utc)
slots: list[PlanningSlot] = []
for i in range(n):
slots.append(
PlanningSlot(
interval_start=base + timedelta(minutes=15 * i),
buy_price=2.0,
sell_price=4.0 + 0.01 * i,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=1000,
ev1_connected=False,
ev2_connected=False,
allow_discharge_export=True,
charge_acquisition_buy_czk_kwh=0.5,
)
)
return slots
def test_budget_scales_with_soc_not_fixed_three(self) -> None:
slots = self._evening_slots(8)
per_slot = 17_000 * 0.95 * 0.25
bat = _battery(uc_wh=64_000.0, min_pct=10.0, max_pct=95.0)
soc_high = 0.92 * bat.soc_max_wh
profitable = set(range(len(slots)))
push_hi = _evening_battery_export_push_indices(
slots,
profitable_export_ts=profitable,
degrad_czk_kwh=0.15,
current_soc_wh=soc_high,
min_soc_wh=bat.min_soc_wh,
soc_max_wh=bat.soc_max_wh,
per_slot_discharge_wh=per_slot,
discharge_slot_buffer=1.5,
)
self.assertGreater(len(push_hi), 3)
soc_low = bat.min_soc_wh + 100.0
push_lo = _evening_battery_export_push_indices(
slots,
profitable_export_ts=profitable,
degrad_czk_kwh=0.15,
current_soc_wh=soc_low,
min_soc_wh=bat.min_soc_wh,
soc_max_wh=bat.soc_max_wh,
per_slot_discharge_wh=per_slot,
discharge_slot_buffer=1.5,
)
self.assertEqual(len(push_lo), 0)
def test_night_window_includes_midnight_excludes_pv_sunrise(self) -> None:
"""23:30 a 00:00 jeden peak; po východu FVE (pv > load) už ne."""
prague = ZoneInfo("Europe/Prague")
slots = [
PlanningSlot(
interval_start=datetime(2026, 5, 25, 23, 15, tzinfo=prague),
buy_price=5.0,
sell_price=3.323,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=1800,
ev1_connected=False,
ev2_connected=False,
allow_discharge_export=True,
),
PlanningSlot(
interval_start=datetime(2026, 5, 25, 23, 30, tzinfo=prague),
buy_price=5.0,
sell_price=3.286,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=1800,
ev1_connected=False,
ev2_connected=False,
allow_discharge_export=True,
),
PlanningSlot(
interval_start=datetime(2026, 5, 26, 0, 0, tzinfo=prague),
buy_price=5.6,
sell_price=3.586,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=1800,
ev1_connected=False,
ev2_connected=False,
allow_discharge_export=True,
),
PlanningSlot(
interval_start=datetime(2026, 5, 26, 6, 0, tzinfo=prague),
buy_price=4.0,
sell_price=3.0,
pv_a_forecast_w=10_000,
pv_b_forecast_w=0,
load_baseline_w=1800,
ev1_connected=False,
ev2_connected=False,
allow_discharge_export=True,
),
]
self.assertTrue(_in_night_battery_export_window(slots[2]))
self.assertFalse(_in_night_battery_export_window(slots[3]))
peak_ts = _evening_peak_export_indices(slots, degrad_czk_kwh=0.15)
self.assertIn(2, peak_ts, "půlnoc musí být v nočním peak pásmu")
self.assertNotIn(3, peak_ts)
bat = _battery(uc_wh=64_000.0, min_pct=12.0, max_pct=95.0)
per_slot = 18_000 * 0.95 * 0.25
push = _evening_battery_export_push_indices(
slots,
profitable_export_ts={0, 1, 2, 3},
degrad_czk_kwh=0.15,
current_soc_wh=0.9 * bat.soc_max_wh,
min_soc_wh=bat.min_soc_wh,
soc_max_wh=bat.soc_max_wh,
per_slot_discharge_wh=per_slot,
discharge_slot_buffer=1.5,
)
self.assertIn(2, push, "nejvyšší sell 00:00 má být v push před 23:30")
self.assertEqual(push[0], 2)
def test_evening_push_budget_matches_r063_formula(self) -> None:
bat = _battery(uc_wh=64_000.0, min_pct=10.0, max_pct=95.0)
soc = 0.85 * bat.soc_max_wh
budget = _evening_push_discharge_budget_wh(
current_soc_wh=soc,
min_soc_wh=bat.min_soc_wh,
soc_max_wh=bat.soc_max_wh,
discharge_slot_buffer=1.5,
)
exportable_full = bat.soc_max_wh - bat.min_soc_wh
available = soc - bat.min_soc_wh
self.assertAlmostEqual(budget, min(available, exportable_full * 1.5))
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:
"""
v25: sell<0 před buy<0 — PV A smí do baterie (bc_pv), ne export za záporný sell.
Curtail PV A (ca) až v okně buy<0 (slot 1).
"""
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)
self.assertGreater(results[0].battery_setpoint_w, 500)
self.assertEqual(results[0].pv_a_curtailed_w, 0)
self.assertGreater(results[1].grid_setpoint_w, 1000)
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:
"""
v25: před buy<0 — SoC u posledního sell≥0 blízko max, před prvním buy<0 pod stropem
(_pre_neg_buy_soc_ceiling_wh), ne kotva na planner floor před sell<0.
"""
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",
)
last_pos = 1
pre_buy = 2
self.assertGreaterEqual(
results[last_pos].battery_soc_target or 0,
60.0,
msg="poslední sell≥0 před buy<0: směr k plné baterii (bez exportu)",
)
self.assertLessEqual(
results[pre_buy].battery_soc_target or 100.0,
75.0,
msg="slot před buy<0: rezerva pro import v buy<0 okně",
)
def test_anchor_uses_planner_floor_even_without_extreme_buy(self) -> None:
"""
v25: bez buy<0 v horizontu — žádný strop před buy<0; poslední sell≥0 může držet vysoké 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.assertEqual(results[0].grid_setpoint_w, 0)
self.assertEqual(results[0].battery_setpoint_w, 0)
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-28-morning-pv-export-priority-v31")
self.assertGreater(
results[0].battery_setpoint_w,
2_500,
f"první sell<0 slot má nabíjet z PV, 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-28-morning-pv-export-priority-v31")
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-28-morning-pv-export-priority-v31")
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_peak_battery_export_at_site_cap(self) -> None:
"""Nejvyšší večerní sell: výrazný export; levnější večerní sloty bez předčasného vývozu."""
prague = ZoneInfo("Europe/Prague")
base = datetime(2026, 5, 25, 17, 0, tzinfo=prague)
sells = [3.5, 3.7, 4.04, 3.75, 3.8, 3.6]
slots = [
PlanningSlot(
interval_start=base + timedelta(minutes=15 * i),
buy_price=0.8,
sell_price=sell,
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=True,
charge_acquisition_buy_czk_kwh=0.8,
)
for i, sell in enumerate(sells)
]
battery = _battery(uc_wh=64_000.0, terminal_soc_value_factor=0.0)
battery.max_discharge_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=16_000, max_export_power_w=16_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,
[None, None],
vehicles,
0.85 * battery.soc_max_wh,
50.0,
operating_mode="AUTO",
)
self.assertEqual(snap["planner_build_tag"], "2026-05-28-morning-pv-export-priority-v31")
peak_idx = sells.index(4.04)
peak = results[peak_idx]
self.assertIn(peak.export_mode, ("BATTERY_SELL", "PV_SURPLUS"))
self.assertGreater(abs(peak.grid_setpoint_w), 5000)
# v27: ge_bat=0 jen před prvním push slotem, ne u všech sell < peak0.05.
for i, r in enumerate(results):
if i >= peak_idx:
continue
self.assertNotEqual(
r.export_mode,
"BATTERY_SELL",
msg=f"slot {i} sell={sells[i]} must not battery-export before first push",
)
def test_midnight_higher_sell_gets_battery_export(self) -> None:
"""home-01 archetyp: export v 00:00 (vyšší sell), ne jen 23:30."""
prague = ZoneInfo("Europe/Prague")
slots = [
PlanningSlot(
interval_start=datetime(2026, 5, 25, 23, 15, tzinfo=prague),
buy_price=5.28,
sell_price=3.323,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=1800,
ev1_connected=False,
ev2_connected=False,
allow_charge=False,
allow_discharge_export=True,
charge_acquisition_buy_czk_kwh=0.8,
),
PlanningSlot(
interval_start=datetime(2026, 5, 25, 23, 30, tzinfo=prague),
buy_price=5.23,
sell_price=3.286,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=1800,
ev1_connected=False,
ev2_connected=False,
allow_charge=False,
allow_discharge_export=True,
charge_acquisition_buy_czk_kwh=0.8,
),
PlanningSlot(
interval_start=datetime(2026, 5, 26, 0, 0, tzinfo=prague),
buy_price=5.63,
sell_price=3.586,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=1800,
ev1_connected=False,
ev2_connected=False,
allow_charge=False,
allow_discharge_export=True,
charge_acquisition_buy_czk_kwh=0.8,
),
]
battery = _battery(uc_wh=64_000.0, terminal_soc_value_factor=0.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, _, snap = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
0.9 * battery.soc_max_wh,
50.0,
operating_mode="AUTO",
)
self.assertEqual(snap["planner_build_tag"], "2026-05-28-morning-pv-export-priority-v31")
r_midnight = results[2]
self.assertEqual(r_midnight.export_mode, "BATTERY_SELL")
self.assertGreaterEqual(abs(r_midnight.grid_setpoint_w), 12_500)
def test_evening_push_export_near_site_cap_home01(self) -> None:
"""home-01 večer: export ≈ min(13.5 kW, 18 kW load), ne (maxload)/2."""
prague = ZoneInfo("Europe/Prague")
base = datetime(2026, 5, 25, 18, 45, tzinfo=prague)
slots = [
PlanningSlot(
interval_start=base,
buy_price=7.3,
sell_price=4.4,
pv_a_forecast_w=0,
pv_b_forecast_w=0,
load_baseline_w=1797,
ev1_connected=False,
ev2_connected=False,
allow_charge=False,
allow_discharge_export=True,
charge_acquisition_buy_czk_kwh=0.8,
)
]
battery = _battery(uc_wh=64_000.0, terminal_soc_value_factor=0.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, _ms, snap = solve_dispatch(
slots,
battery,
hp,
grid,
[None, None],
vehicles,
0.9 * battery.soc_max_wh,
50.0,
operating_mode="AUTO",
)
self.assertEqual(snap["planner_build_tag"], "2026-05-28-morning-pv-export-priority-v31")
r = results[0]
self.assertEqual(r.export_mode, "BATTERY_SELL")
self.assertGreaterEqual(abs(r.grid_setpoint_w), 12_500)
self.assertLessEqual(abs(r.grid_setpoint_w), 13_500)
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: spot sell<0 → nabít/vent B; sell>=0 → LP volí export vs bc (ne tvrdý curtail)."""
def test_morning_pre_neg_discharge_exports_pv_not_full_curtail(self) -> None:
"""07:00 archetyp: sell>0 + PV před buy<0 — FVE do sítě, ne plný ge_bat push + curtail."""
prague = ZoneInfo("Europe/Prague")
base = datetime(2026, 5, 26, 7, 0, tzinfo=prague)
slots: list[PlanningSlot] = []
for i in range(12):
slots.append(
PlanningSlot(
interval_start=base + timedelta(minutes=15 * i),
buy_price=5.9 if i == 0 else (0.9 if i < 8 else -0.5),
sell_price=3.79 if i == 0 else (3.2 if i < 8 else -0.3),
pv_a_forecast_w=629 if i == 0 else (3000 if i < 4 else 8000),
pv_b_forecast_w=0,
load_baseline_w=2000,
ev1_connected=False,
ev2_connected=False,
allow_charge=False,
allow_discharge_export=True,
charge_acquisition_buy_czk_kwh=0.8,
)
)
battery = _battery(uc_wh=64_000.0, min_pct=12.0, arb_pct=20.0)
battery.max_discharge_power_w = 18_000
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=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, _, snap = solve_dispatch(
slots, battery, hp, grid, [None, None], vehicles, 0.55 * battery.soc_max_wh, 50.0, operating_mode="AUTO"
)
self.assertEqual(snap["planner_build_tag"], "2026-05-28-morning-pv-export-priority-v31")
r0 = results[0]
self.assertLess(
r0.pv_a_curtailed_w,
500,
"nesmí useknout celou FVE kvůli plnému ge_bat push (archetyp 07:00)",
)
self.assertNotEqual(r0.export_mode, "BATTERY_SELL")
if r0.grid_setpoint_w < -500:
self.assertEqual(r0.export_mode, "PV_SURPLUS")
def test_positive_sell_full_battery_exports_pv_not_curtail(self) -> None:
"""Odpoledne sell ~3 Kč, večer ~6,6 — plná baterie: export FVE, ne pv_store curtail."""
slots = [
PlanningSlot(
interval_start=datetime(2026, 5, 25, 12, 0, tzinfo=timezone.utc),
buy_price=2.5,
sell_price=3.0,
pv_a_forecast_w=10_000,
pv_b_forecast_w=0,
load_baseline_w=1800,
ev1_connected=False,
ev2_connected=False,
allow_charge=False,
allow_discharge_export=False,
future_sell_opportunity_czk_kwh=6.6,
)
]
battery = _battery(uc_wh=64_000.0, min_pct=12.0, arb_pct=20.0)
battery.max_charge_power_w = 18_000
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=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.98 * battery.soc_max_wh
results, _, _ = solve_dispatch(
slots, battery, hp, grid, [None, None], vehicles, soc0, 50.0, operating_mode="AUTO"
)
r = results[0]
self.assertLess(r.grid_setpoint_w, -500, "přebytek FVE do sítě při kladném sell")
self.assertLess(r.pv_a_curtailed_w, 5000, "nesmí useknout celé pole A kvůli pv_store")
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.assertLessEqual(
r.export_limit_w,
13_500,
msg="export_limit_w odpovídá site limitu",
)
self.assertLessEqual(abs(r.battery_setpoint_w), 18_000)
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()