"""MILP dispatch: dvouúrovňové SoC a záporná nákupní cena (bez DB).""" from __future__ import annotations import unittest from datetime import datetime, timedelta, timezone from types import SimpleNamespace from services.planning_engine import ( DispatchResult, PlanningSlot, _dynamic_arb_floor_wh_series, _dispatch_result_comparison, _prewindow_deferral_slots, _slots_until_buy_le_threshold, _slots_until_sell_lt, _soc_panel_min_wh_series, solve_dispatch, solve_dispatch_two_pass, ) def _slot( *, load: int = 2000, buy: float = 3.0, sell: float = 3.0, pv_a: int = 0, pv_b: int = 0, ) -> PlanningSlot: return PlanningSlot( interval_start=datetime(2026, 4, 3, 12, 0, tzinfo=timezone.utc), buy_price=buy, sell_price=sell, pv_a_forecast_w=pv_a, pv_b_forecast_w=pv_b, load_baseline_w=load, ev1_connected=False, ev2_connected=False, is_predicted_price=False, ) def _battery( *, uc_wh: float = 100_000.0, min_pct: float = 10.0, arb_pct: float = 20.0, max_pct: float = 95.0, terminal_soc_value_factor: float = 0.9, ) -> SimpleNamespace: uc = uc_wh min_wh = min_pct / 100.0 * uc arb_wh = arb_pct / 100.0 * uc return SimpleNamespace( usable_capacity_wh=uc, min_soc_wh=min_wh, arb_floor_wh=arb_wh, reserve_soc_wh=arb_wh, soc_max_wh=max_pct / 100.0 * uc, charge_efficiency=0.95, discharge_efficiency=0.95, degradation_cost_czk_kwh=0.15, max_charge_power_w=10_000, max_discharge_power_w=10_000, planner_terminal_soc_value_factor=terminal_soc_value_factor, ) class SlotsUntilSellNegativeTests(unittest.TestCase): def test_slots_until_first_negative_sell(self) -> None: base = datetime(2026, 4, 3, 0, 0, tzinfo=timezone.utc) slots: list[PlanningSlot] = [] for i in range(10): slots.append( PlanningSlot( interval_start=base + timedelta(minutes=15 * i), buy_price=1.0, sell_price=2.0 if i < 4 else -0.5, pv_a_forecast_w=0, pv_b_forecast_w=0, load_baseline_w=500, ev1_connected=False, ev2_connected=False, ) ) dist = _slots_until_sell_lt(slots, 0.0) self.assertEqual(dist[0], 4) self.assertEqual(dist[3], 1) self.assertEqual(dist[4], 0) def test_prewindow_deferral_prefers_sell_anchor(self) -> None: """Když existuje záporný prodej, kotva je vzdálenost k němu, ne k extrémnímu buy.""" base = datetime(2026, 4, 3, 0, 0, tzinfo=timezone.utc) slots: list[PlanningSlot] = [] for i in range(8): slots.append( PlanningSlot( interval_start=base + timedelta(minutes=15 * i), buy_price=-50.0, sell_price=1.0 if i < 2 else -0.1, pv_a_forecast_w=0, pv_b_forecast_w=0, load_baseline_w=500, ev1_connected=False, ev2_connected=False, ) ) adv = _prewindow_deferral_slots(slots, -2.0) self.assertEqual(adv[0], 2) def test_prewindow_deferral_falls_back_to_buy_when_no_negative_sell(self) -> None: base = datetime(2026, 4, 3, 0, 0, tzinfo=timezone.utc) slots: list[PlanningSlot] = [] for i in range(10): slots.append( PlanningSlot( interval_start=base + timedelta(minutes=15 * i), buy_price=3.0 if i < 7 else -10.0, sell_price=2.0, pv_a_forecast_w=0, pv_b_forecast_w=0, load_baseline_w=500, ev1_connected=False, ev2_connected=False, ) ) adv = _prewindow_deferral_slots(slots, -2.0) self.assertEqual(adv[0], 7) class SlotsUntilBuyExtremeTests(unittest.TestCase): def test_slots_until_first_extreme(self) -> None: base = datetime(2026, 4, 3, 0, 0, tzinfo=timezone.utc) slots: list[PlanningSlot] = [] for i in range(10): slots.append( PlanningSlot( interval_start=base + timedelta(minutes=15 * i), buy_price=1.0, sell_price=1.0, pv_a_forecast_w=0, pv_b_forecast_w=0, load_baseline_w=500, ev1_connected=False, ev2_connected=False, ) ) slots[-1] = PlanningSlot( interval_start=slots[-1].interval_start, buy_price=-10.0, sell_price=0.0, pv_a_forecast_w=0, pv_b_forecast_w=0, load_baseline_w=500, ev1_connected=False, ev2_connected=False, ) dist = _slots_until_buy_le_threshold(slots, -2.0) self.assertEqual(dist[0], 9) self.assertEqual(dist[8], 1) self.assertEqual(dist[9], 0) def test_prewindow_clamps_relaxed_floor_until_close(self) -> None: sm = [5000.0] * 10 dist = [9, 8, 7, 6, 5, 4, 3, 2, 1, 0] # obecná kotva (sell nebo buy) panel = _soc_panel_min_wh_series(sm, dist, 10_000.0, 20_000.0, 2) self.assertEqual(panel[0], 20_000.0) self.assertEqual(panel[6], 20_000.0) self.assertEqual(panel[7], 5000.0) self.assertEqual(panel[9], 5000.0) class DynamicArbFloorTests(unittest.TestCase): def test_more_pv_ahead_lowers_floor(self) -> None: """Čím víc FVE ve lookahead, tím nižší ekonomická podlaha v prvním slotu.""" min_w = 1_000.0 base_w = 2_000.0 uc = 10_000.0 s0 = _slot() s_low_pv = replace_slot(s0, pv_a=100, pv_b=0) s_high_pv = replace_slot(s0, pv_a=50_000, pv_b=0) ser_low = _dynamic_arb_floor_wh_series([s_low_pv] * 40, min_w, base_w, uc) ser_high = _dynamic_arb_floor_wh_series([s_high_pv] * 40, min_w, base_w, uc) self.assertLess(ser_high[0], ser_low[0]) self.assertGreaterEqual(ser_low[0], min_w) self.assertLessEqual(ser_low[0], base_w) def replace_slot( s: PlanningSlot, *, pv_a: int | None = None, pv_b: int | None = None, load: int | None = None, ) -> PlanningSlot: return PlanningSlot( interval_start=s.interval_start, buy_price=s.buy_price, sell_price=s.sell_price, pv_a_forecast_w=pv_a if pv_a is not None else s.pv_a_forecast_w, pv_b_forecast_w=pv_b if pv_b is not None else s.pv_b_forecast_w, load_baseline_w=load if load is not None else s.load_baseline_w, ev1_connected=s.ev1_connected, ev2_connected=s.ev2_connected, is_predicted_price=s.is_predicted_price, ) class PlanningDispatchMilpTests(unittest.TestCase): def test_dispatch_result_comparison_marks_changed_slots(self) -> None: dt = datetime(2026, 4, 3, 12, 0, tzinfo=timezone.utc) active = [ DispatchResult( interval_start=dt, battery_setpoint_w=1000, battery_soc_target=50.0, grid_setpoint_w=0, export_limit_w=0, export_mode="NONE", deye_physical_mode="PASSIVE", deye_gen_cutoff_enabled=False, ev1_setpoint_w=None, ev2_setpoint_w=None, ev1_via_bat_w=0, ev2_via_bat_w=0, heat_pump_enabled=False, heat_pump_setpoint_w=0, pv_a_curtailed_w=0, expected_cost_czk=1.0, effective_buy_price=1.0, effective_sell_price=1.0, is_predicted_price=False, ) ] 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, ) ] cmp = _dispatch_result_comparison(active, 10, "v1", peer, 12, "v2") self.assertEqual(cmp["active"]["planner_version"], "v1") self.assertEqual(cmp["peer"]["planner_version"], "v2") self.assertEqual(cmp["diff"]["changed_slots"], 1) self.assertEqual(len(cmp["slot_diffs"]), 1) def test_planner_version_is_recorded_in_snapshot(self) -> None: slots = [_slot(load=500, buy=1.0, sell=1.0, pv_a=0, pv_b=0) for _ in range(2)] battery = _battery() hp = SimpleNamespace(rated_heating_power_w=0, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0) grid = SimpleNamespace(max_import_power_w=20_000, max_export_power_w=20_000) vehicles = [ SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0), SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0), ] results, _ms, snap = solve_dispatch( slots, battery, hp, grid, [], vehicles, current_soc_wh=0.5 * battery.usable_capacity_wh, current_tuv_temp_c=50.0, planner_version="v2", ) self.assertEqual(len(results), 2) self.assertEqual(snap["inputs"]["planner_version"], "v2") def test_neg_sell_with_future_neg_buy_prefers_curtail_pv_a_over_export(self) -> None: """ Když: - aktuální slot má sell < 0 (export je náklad), - v horizontu existuje budoucí buy < 0, - a zároveň existuje PV B (necurtailable) někde v horizontu, solver preferuje curtail PV A (ca) místo placeného exportu ge. """ slots = [ _slot(load=0, buy=3.0, sell=-0.1, pv_a=5000, pv_b=0), _slot(load=0, buy=-10.0, sell=1.0, pv_a=0, pv_b=5000), ] battery = _battery(uc_wh=50_000.0) hp = SimpleNamespace( rated_heating_power_w=0, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0, ) grid = SimpleNamespace(max_import_power_w=20_000, max_export_power_w=20_000) vehicles = [ SimpleNamespace( max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0, ), SimpleNamespace( max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0, ), ] soc0 = 0.50 * battery.usable_capacity_wh results, _ms, _ = solve_dispatch( slots, battery, hp, grid, [None, None], vehicles, soc0, 50.0, tuv_delta_stats=None, operating_mode="AUTO", ) self.assertEqual(len(results), 2) # Slot 0: záporný sell — žádný export FVE do sítě (LP guard sell < acquisition). self.assertNotEqual(results[0].export_mode, "PV_SURPLUS") self.assertNotEqual(results[0].export_mode, "PV_SURPLUS") def test_pv_surplus_export_uses_hard_export_cap(self) -> None: slots = [ PlanningSlot( interval_start=datetime(2026, 4, 3, 12, 0, tzinfo=timezone.utc), buy_price=3.0, sell_price=2.5, pv_a_forecast_w=20_000, pv_b_forecast_w=0, load_baseline_w=0, ev1_connected=False, ev2_connected=False, is_predicted_price=False, allow_charge=False, allow_discharge_export=False, ), ] battery = _battery() hp = SimpleNamespace( rated_heating_power_w=0, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0, ) grid = SimpleNamespace(max_import_power_w=20_000, max_export_power_w=13_500) vehicles = [ SimpleNamespace( max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0, ), SimpleNamespace( max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0, ), ] soc0 = battery.soc_max_wh results, _ms, _ = solve_dispatch( slots, battery, hp, grid, [None, None], vehicles, soc0, 50.0, tuv_delta_stats=None, operating_mode="AUTO", ) self.assertEqual(len(results), 1) self.assertEqual(results[0].export_mode, "PV_SURPLUS") self.assertEqual(results[0].export_limit_w, 13_500) self.assertGreater(results[0].pv_a_curtailed_w, 0) def test_two_tier_soc_solves_optimal(self) -> None: slots = [_slot()] battery = _battery() hp = SimpleNamespace( rated_heating_power_w=0, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0, ) grid = SimpleNamespace(max_import_power_w=15_000, max_export_power_w=15_000) vehicles = [ SimpleNamespace( max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0, ), SimpleNamespace( max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0, ), ] soc0 = 0.15 * battery.usable_capacity_wh results, ms, _ = solve_dispatch( slots, battery, hp, grid, [None, None], vehicles, soc0, 50.0, tuv_delta_stats=None, operating_mode="AUTO", ) self.assertGreaterEqual(ms, 0) self.assertEqual(len(results), 1) def test_deep_discharge_allows_covering_load_only(self) -> None: slots = [ _slot(load=3000, buy=1.0, sell=6.0, pv_a=0, pv_b=0), _slot(load=3000, buy=1.0, sell=6.0, pv_a=0, pv_b=0), ] battery = _battery(uc_wh=50_000.0) hp = SimpleNamespace( rated_heating_power_w=0, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0, ) grid = SimpleNamespace(max_import_power_w=20_000, max_export_power_w=20_000) vehicles = [ SimpleNamespace( max_charge_power_w=11_000, battery_capacity_kwh=50.0, default_target_soc_pct=80.0, ), SimpleNamespace( max_charge_power_w=11_000, battery_capacity_kwh=50.0, default_target_soc_pct=80.0, ), ] soc0 = 0.12 * battery.usable_capacity_wh results, _ms, _ = solve_dispatch( slots, battery, hp, grid, [None, None], vehicles, soc0, 50.0, tuv_delta_stats=None, operating_mode="AUTO", ) self.assertEqual(len(results), 2) def test_negative_buy_price_allows_import_for_baseline(self) -> None: slots = [_slot(load=6000, buy=-0.5, sell=2.0)] battery = _battery() hp = SimpleNamespace( rated_heating_power_w=8000, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0, ) grid = SimpleNamespace(max_import_power_w=25_000, max_export_power_w=15_000) vehicles = [ SimpleNamespace( max_charge_power_w=11_000, battery_capacity_kwh=50.0, default_target_soc_pct=80.0, ), SimpleNamespace( max_charge_power_w=11_000, battery_capacity_kwh=50.0, default_target_soc_pct=80.0, ), ] soc0 = 0.5 * battery.usable_capacity_wh results, _ms, _ = solve_dispatch( slots, battery, hp, grid, [None, None], vehicles, soc0, 50.0, tuv_delta_stats=None, operating_mode="AUTO", ) self.assertGreaterEqual(results[0].grid_setpoint_w, 0) def test_export_implies_end_soc_at_least_reserve(self) -> None: """Bez arbitrážní relaxace: při ge >= 1 W musí koncové soc[t] >= arb_base_wh (rezerva z DB).""" slots = [ _slot(load=500, buy=2.0, sell=8.0, pv_a=0, pv_b=0), _slot(load=500, buy=2.0, sell=8.0, pv_a=0, pv_b=0), ] battery = _battery(uc_wh=100_000.0, min_pct=10.0, arb_pct=20.0) hp = SimpleNamespace( rated_heating_power_w=0, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0, ) grid = SimpleNamespace(max_import_power_w=50_000, max_export_power_w=50_000) vehicles = [ SimpleNamespace( max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0, ), SimpleNamespace( max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0, ), ] soc0 = 0.22 * battery.usable_capacity_wh results, _ms, _ = solve_dispatch( slots, battery, hp, grid, [None, None], vehicles, soc0, 50.0, tuv_delta_stats=None, operating_mode="AUTO", ) reserve_pct = 20.0 for r in results: if r.grid_setpoint_w < 0: self.assertGreaterEqual( r.battery_soc_target, reserve_pct - 0.2, msg="export slot must end at or above reserve SoC", ) def test_export_before_extreme_negative_buy_can_end_below_reserve(self) -> None: """ Při relaxovaném soc_min (záporný buy v lookahead) smí významný export skončit u planner floor, ne u provozní rezervy — jinak nejde ráno vypustit do sítě a nachystat kapacitu před levným nákupem. """ base = datetime(2026, 4, 3, 6, 0, tzinfo=timezone.utc) s0 = PlanningSlot( interval_start=base, buy_price=2.5, sell_price=2.0, pv_a_forecast_w=0, pv_b_forecast_w=0, load_baseline_w=400, ev1_connected=False, ev2_connected=False, is_predicted_price=False, allow_charge=True, allow_discharge_export=True, ) s1 = PlanningSlot( interval_start=base + timedelta(minutes=15), buy_price=-12.0, sell_price=-0.5, pv_a_forecast_w=0, pv_b_forecast_w=0, load_baseline_w=400, ev1_connected=False, ev2_connected=False, is_predicted_price=False, allow_charge=True, allow_discharge_export=True, ) slots = [s0, s1] battery = _battery(uc_wh=10_000.0, min_pct=10.0, arb_pct=20.0) battery.planner_extreme_buy_threshold_czk_kwh = -2.0 battery.planner_discharge_floor_percent = 5.0 battery.max_charge_power_w = 50_000 battery.max_discharge_power_w = 50_000 hp = SimpleNamespace( rated_heating_power_w=0, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0, ) grid = SimpleNamespace(max_import_power_w=50_000, max_export_power_w=50_000) vehicles = [ SimpleNamespace( max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0, ), SimpleNamespace( max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0, ), ] soc0 = 0.88 * battery.usable_capacity_wh results, _ms, _ = solve_dispatch( slots, battery, hp, grid, [None, None], vehicles, soc0, 50.0, tuv_delta_stats=None, operating_mode="AUTO", ) self.assertEqual(len(results), 2) if results[0].grid_setpoint_w < 0: self.assertLess( results[0].battery_soc_target, 19.0, msg="with relaxed soc_min, first-slot export should be able to finish below reserve %", ) def test_negative_sell_forbids_battery_export_arbitrage(self) -> None: """ Pokud sell < 0, solver nesmí vybíjet baterii do sítě pro arbitráž (dump musí proběhnout předtím). V okně sell<0 smí export vzniknout jen z přebytku FVE; zde ale FVE=0, takže očekáváme grid_setpoint>=0. """ base = datetime(2026, 4, 3, 6, 0, tzinfo=timezone.utc) s0 = PlanningSlot( interval_start=base, buy_price=2.0, sell_price=2.0, pv_a_forecast_w=0, pv_b_forecast_w=0, load_baseline_w=0, ev1_connected=False, ev2_connected=False, is_predicted_price=False, allow_charge=True, allow_discharge_export=True, ) s1 = PlanningSlot( interval_start=base + timedelta(minutes=15), buy_price=2.0, sell_price=-0.5, pv_a_forecast_w=0, pv_b_forecast_w=0, load_baseline_w=0, ev1_connected=False, ev2_connected=False, is_predicted_price=False, allow_charge=True, allow_discharge_export=True, ) s2 = PlanningSlot( interval_start=base + timedelta(minutes=30), buy_price=-15.0, sell_price=-1.0, pv_a_forecast_w=0, pv_b_forecast_w=0, load_baseline_w=0, ev1_connected=False, ev2_connected=False, is_predicted_price=False, allow_charge=True, allow_discharge_export=True, ) slots = [s0, s1, s2] battery = _battery(uc_wh=10_000.0, min_pct=10.0, arb_pct=20.0) battery.planner_extreme_buy_threshold_czk_kwh = -2.0 battery.planner_discharge_floor_percent = 5.0 battery.max_charge_power_w = 50_000 battery.max_discharge_power_w = 50_000 hp = SimpleNamespace( rated_heating_power_w=0, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0, ) grid = SimpleNamespace(max_import_power_w=50_000, max_export_power_w=50_000) vehicles = [ SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0), SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0), ] soc0 = 0.9 * battery.usable_capacity_wh results, _ms, _ = solve_dispatch( slots, battery, hp, grid, [None, None], vehicles, soc0, 50.0, tuv_delta_stats=None, operating_mode="AUTO", ) self.assertEqual(len(results), 3) # V sell<0 slotu bez FVE a bez zátěže nesmí být export (to by muselo být z baterie). self.assertGreaterEqual(results[1].grid_setpoint_w, 0) # A zároveň nesmí být baterie ve výboji (dump musí proběhnout předtím). self.assertGreaterEqual(results[1].battery_setpoint_w, 0) def test_anchor_hits_floor_before_first_negative_sell(self) -> None: """ Pokud se v horizontu objeví první sell<0 a současně existuje planner floor (relaxace), solver má skončit už v předchozím slotu u planner floor (cca 5 %), ne na ~15 %. """ base = datetime(2026, 4, 3, 6, 0, tzinfo=timezone.utc) # Slot 0-1: sell >= 0; slot 2: první sell < 0; slot 3: extrémně záporný buy (motivace k bufferu). slots = [ PlanningSlot( interval_start=base, buy_price=3.0, sell_price=1.0, pv_a_forecast_w=0, pv_b_forecast_w=0, load_baseline_w=0, ev1_connected=False, ev2_connected=False, allow_charge=True, allow_discharge_export=True, ), PlanningSlot( interval_start=base + timedelta(minutes=15), buy_price=3.0, sell_price=0.5, pv_a_forecast_w=0, pv_b_forecast_w=0, load_baseline_w=0, ev1_connected=False, ev2_connected=False, allow_charge=True, allow_discharge_export=True, ), PlanningSlot( interval_start=base + timedelta(minutes=30), buy_price=3.0, sell_price=-0.2, pv_a_forecast_w=0, pv_b_forecast_w=0, load_baseline_w=0, ev1_connected=False, ev2_connected=False, allow_charge=True, allow_discharge_export=True, ), PlanningSlot( interval_start=base + timedelta(minutes=45), buy_price=-20.0, sell_price=-1.0, pv_a_forecast_w=0, pv_b_forecast_w=0, load_baseline_w=0, ev1_connected=False, ev2_connected=False, allow_charge=True, allow_discharge_export=True, ), ] battery = _battery(uc_wh=20_000.0, min_pct=10.0, arb_pct=20.0) battery.planner_extreme_buy_threshold_czk_kwh = -2.0 battery.planner_discharge_floor_percent = 5.0 battery.max_charge_power_w = 50_000 battery.max_discharge_power_w = 50_000 hp = SimpleNamespace(rated_heating_power_w=0, tuv_min_temp_c=45.0, tuv_target_temp_c=55.0) grid = SimpleNamespace(max_import_power_w=50_000, max_export_power_w=50_000) vehicles = [ SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0), SimpleNamespace(max_charge_power_w=0, battery_capacity_kwh=1.0, default_target_soc_pct=80.0), ] soc0 = 0.9 * battery.usable_capacity_wh results, _ms, _ = solve_dispatch( slots, battery, hp, grid, [None, None], vehicles, soc0, 50.0, tuv_delta_stats=None, operating_mode="AUTO", ) # Slot index 1 je poslední před prvním sell<0 (index 2). self.assertLessEqual( results[1].battery_soc_target, 6.0, msg="anchor should drive SoC close to planner floor before first negative sell", ) def test_anchor_uses_planner_floor_even_without_extreme_buy(self) -> None: """ Regrese: pokud v horizontu není buy <= threshold (soc_min_series by se nerelaxovala), kotva před sell<0 má stejně mířit na planner floor (5 %), ne na base min SoC. """ base = datetime(2026, 4, 3, 6, 0, tzinfo=timezone.utc) slots = [ PlanningSlot( interval_start=base, buy_price=3.0, sell_price=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, ), ] 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", ) # Slot index 1 je poslední před prvním sell<0 (index 2). self.assertLessEqual(results[1].battery_soc_target, 6.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 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.assertGreaterEqual(results[0].grid_setpoint_w, 0) self.assertEqual(results[0].deye_physical_mode, "PASSIVE") 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 + 500) self.assertNotEqual(morning.export_mode, "PV_SURPLUS") self.assertLessEqual(vt_before_nt.grid_setpoint_w, 4_000) self.assertLessEqual(vt_before_nt.battery_setpoint_w, 2_000) class ChargeAcquisitionArbitrageTests(unittest.TestCase): """Mezi-slotová arbitráž: večerní export při nízké charge_acquisition z SQL.""" def test_evening_battery_export_when_sell_above_acquisition(self) -> None: base = datetime(2026, 5, 21, 10, 0, tzinfo=timezone.utc) cheap = (0.75, 0.25) peak = (7.0, 4.8) slots: list[PlanningSlot] = [] for i in range(6): buy, sell = cheap if i < 2 else peak slots.append( PlanningSlot( interval_start=base + timedelta(minutes=15 * i), buy_price=buy, sell_price=sell, pv_a_forecast_w=0, pv_b_forecast_w=0, load_baseline_w=800, ev1_connected=False, ev2_connected=False, is_predicted_price=False, allow_charge=i < 2, allow_discharge_export=i >= 2, charge_acquisition_buy_czk_kwh=0.75, charge_acquisition_cutoff_at=base + timedelta(minutes=30), ) ) battery = _battery(uc_wh=64_000.0, terminal_soc_value_factor=0.5) 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] self.assertLess(evening.grid_setpoint_w, -5_000) self.assertEqual(evening.export_mode, "BATTERY_SELL") inputs = snap.get("inputs") or {} self.assertTrue(inputs.get("two_pass_enabled")) 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") if __name__ == "__main__": unittest.main()