dalsi fixy

backend/tests/test_planning_charge_slot_selection.py

@@ -3,8 +3,8 @@
 Logika je v DB: ems.fn_load_planning_slots_full. Kopie algoritmu pro unit testy bez PG.
 
 Charge mask:
-  A) PV-surplus: store_score DESC, dokud PV nepokryje charge target.
-  B) Non-PV: AM/PM, OTE-first, buy≤ref+degrad, lookahead, cap 6 slotů.
+  B) Grid ze sítě první: AM/PM 50/50 Wh, buy≤min(buy v pásmu)+band, i s FVE.
+  A) PV-surplus: store_score DESC, doplní zbytek po vrstvě B.
 
 Discharge-export mask:
   ref_buy = min(buy) celého horizontu.
@@ -22,9 +22,9 @@ from services.planning_engine import INTERVAL_H, PlanningSlot
 
 _PRAGUE = ZoneInfo("Europe/Prague")
 _LOOKAHEAD_SLOTS = 4
-_GRID_CHARGE_CAP_AM = 6
-_GRID_CHARGE_CAP_PM = 6
 _BUY_LOOKAHEAD_EPS = 0.05
+_BUY_CHARGE_BAND = 0.40
+_MAX_GRID_CHARGE_CAP = 24
 
 
 def _future_sell(slots: list[PlanningSlot], t: int) -> float:
@@ -66,7 +66,14 @@ def _select_charge_slots(
 
     reserve_wh = float(getattr(battery, "reserve_soc_wh", 0) or 0)
     degrad = float(getattr(battery, "degradation_cost_czk_kwh", 0.15) or 0.15)
-    ref_buy = min(float(s.buy_price) for s in slots)
+    ref_buy_am = min(
+        (float(s.buy_price) for s in slots if _prague_hour(s) < 12),
+        default=min(float(s.buy_price) for s in slots),
+    )
+    ref_buy_pm = min(
+        (float(s.buy_price) for s in slots if _prague_hour(s) >= 12),
+        default=min(float(s.buy_price) for s in slots),
+    )
 
     eta = float(getattr(battery, "charge_efficiency", 1.0) or 1.0)
     max_p_w = float(getattr(battery, "max_charge_power_w", 0.0) or 0.0)
@@ -92,8 +99,63 @@ def _select_charge_slots(
         chg_pm = charge_target_wh - chg_am
 
     selected: set[int] = set()
+    grid_filled_wh = 0.0
 
-    # A) PV-surplus: highest store_score first
+    cap_am = (
+        max(1, min(_MAX_GRID_CHARGE_CAP, int(chg_am / per_slot_full_wh) + 1))
+        if per_slot_full_wh > 0
+        else 6
+    )
+    cap_pm = (
+        max(1, min(_MAX_GRID_CHARGE_CAP, int(chg_pm / per_slot_full_wh) + 1))
+        if per_slot_full_wh > 0
+        else 6
+    )
+
+    def _grid_b_ok(t: int, ref_buy_seg: float) -> bool:
+        s = slots[t]
+        buy = float(s.buy_price)
+        if buy > ref_buy_seg + _BUY_CHARGE_BAND:
+            return False
+        nxt = _buy_min_next_n(slots, t)
+        if nxt is not None and buy > nxt + _BUY_LOOKAHEAD_EPS:
+            return False
+        return True
+
+    if purchase_pricing_mode != "fixed":
+        am_candidates = [
+            (t, getattr(slots[t], "is_predicted_price", False), float(slots[t].buy_price))
+            for t in range(len(slots))
+            if _grid_b_ok(t, ref_buy_am) and _prague_hour(slots[t]) < 12
+        ]
+        am_candidates.sort(key=lambda x: (int(x[1]), x[2], x[0]))
+        cum = 0.0
+        grid_am = 0
+        for t, _pred, _price in am_candidates:
+            if cum >= chg_am or per_slot_full_wh <= 0 or grid_am >= cap_am:
+                break
+            selected.add(t)
+            cum += per_slot_full_wh
+            grid_am += 1
+        grid_filled_wh += cum
+
+        pm_candidates = [
+            (t, getattr(slots[t], "is_predicted_price", False), float(slots[t].buy_price))
+            for t in range(len(slots))
+            if _grid_b_ok(t, ref_buy_pm) and _prague_hour(slots[t]) >= 12
+        ]
+        pm_candidates.sort(key=lambda x: (int(x[1]), x[2], x[0]))
+        cum = 0.0
+        grid_pm = 0
+        for t, _pred, _price in pm_candidates:
+            if cum >= chg_pm or per_slot_full_wh <= 0 or grid_pm >= cap_pm:
+                break
+            selected.add(t)
+            cum += per_slot_full_wh
+            grid_pm += 1
+        grid_filled_wh += cum
+
+    pv_layer_cap = max(charge_target_wh - grid_filled_wh, 0.0)
     pv_candidates: list[tuple[int, float, float]] = []
     for t, s in enumerate(slots):
         pv_surplus_w = max(0, s.pv_a_forecast_w + s.pv_b_forecast_w - s.load_baseline_w)
@@ -103,58 +165,11 @@ def _select_charge_slots(
     pv_candidates.sort(key=lambda x: (-x[1], x[0]))
     cum = 0.0
     for t, _score, pv_surplus_w in pv_candidates:
-        if cum >= charge_target_wh:
+        if cum >= pv_layer_cap:
             break
         selected.add(t)
         cum += min(pv_surplus_w, max_p_w) * eta * INTERVAL_H
 
-    if purchase_pricing_mode == "fixed":
-        return selected
-
-    def _grid_b_ok(t: int) -> bool:
-        s = slots[t]
-        if max(0, s.pv_a_forecast_w + s.pv_b_forecast_w - s.load_baseline_w) > 0:
-            return False
-        buy = float(s.buy_price)
-        sell = float(s.sell_price)
-        if buy > ref_buy + degrad:
-            return False
-        nxt = _buy_min_next_n(slots, t)
-        if nxt is not None and buy > nxt + _BUY_LOOKAHEAD_EPS:
-            return False
-        return True
-
-    # B) AM
-    am_candidates = [
-        (t, getattr(slots[t], "is_predicted_price", False), float(slots[t].buy_price))
-        for t in range(len(slots))
-        if t not in selected and _grid_b_ok(t) and _prague_hour(slots[t]) < 12
-    ]
-    am_candidates.sort(key=lambda x: (int(x[1]), x[2], x[0]))
-    cum = 0.0
-    grid_am = 0
-    for t, _pred, _price in am_candidates:
-        if cum >= chg_am or per_slot_full_wh <= 0 or grid_am >= _GRID_CHARGE_CAP_AM:
-            break
-        selected.add(t)
-        cum += per_slot_full_wh
-        grid_am += 1
-
-    pm_candidates = [
-        (t, getattr(slots[t], "is_predicted_price", False), float(slots[t].buy_price))
-        for t in range(len(slots))
-        if t not in selected and _grid_b_ok(t) and _prague_hour(slots[t]) >= 12
-    ]
-    pm_candidates.sort(key=lambda x: (int(x[1]), x[2], x[0]))
-    cum = 0.0
-    grid_pm = 0
-    for t, _pred, _price in pm_candidates:
-        if cum >= chg_pm or per_slot_full_wh <= 0 or grid_pm >= _GRID_CHARGE_CAP_PM:
-            break
-        selected.add(t)
-        cum += per_slot_full_wh
-        grid_pm += 1
-
     return selected
 
 
@@ -311,6 +326,49 @@ class SelectChargeSlotsTests(unittest.TestCase):
         out = _select_charge_slots(slots, battery, current_soc_wh=0.0)
         self.assertIn(2, out, "Nejlevnější buy v horizontu (PM) musí být vybrán")
 
+    def test_cheap_pm_with_pv_surplus_gets_grid_charge(self) -> None:
+        """Regrese home-01: levné PM VT (~0,8) i s FVE musí projít grid maskou B."""
+        base = datetime(2026, 5, 21, 10, 0, tzinfo=timezone.utc)
+        slots = [
+            _slot(
+                buy=0.80,
+                sell=-0.08,
+                pv=2_500,
+                load=3_400,
+                interval_start=base,
+            ),
+            _slot(
+                buy=0.72,
+                sell=-0.13,
+                pv=500,
+                load=3_400,
+                interval_start=base + timedelta(minutes=15),
+            ),
+            _slot(
+                buy=2.50,
+                sell=1.40,
+                pv=2_000,
+                load=3_800,
+                interval_start=base + timedelta(hours=5),
+            ),
+            _slot(
+                buy=5.50,
+                sell=3.80,
+                pv=100,
+                load=2_900,
+                interval_start=base + timedelta(hours=9),
+            ),
+        ]
+        battery = _battery(uc_wh=64_000.0)
+        soc = 0.46 * battery.usable_capacity_wh
+        out = _select_charge_slots(slots, battery, current_soc_wh=soc)
+        self.assertIn(1, out, "Levnější PM slot (lookahead) má allow_charge i s FVE")
+        self.assertNotIn(
+            2,
+            out,
+            "Drahý odpolední slot nemá být v grid maskě B jen kvůli globálnímu min",
+        )
+
     def test_vt_before_nt_skips_expensive_pm_slot(self) -> None:
         """Regrese home-01: 12:45 VT drahý, za 15 min NT levný → PM grid charge ne v 12:45."""
         base = datetime(2026, 5, 21, 10, 45, tzinfo=timezone.utc)