Use observed SoC for neg-prep cushion and evening drain (v40).

Pre-neg forecast cushion and evening push before negative-sell days now use telemetry SoC instead of chaining LP targets across days, so the planner does not stop discharging early when BMS is higher than the model.

Co-authored-by: Cursor <cursoragent@cursor.com>

backend/services/planning_engine.py

@@ -71,7 +71,7 @@ NEG_BUY_CHARGE_SHORTFALL_PENALTY_CZK_KWH = 100.0
 PRE_NEG_CHARGE_PENALTY_CZK_KWH = 400.0
 PRE_NEG_BATT_EXPORT_SHORTFALL_PENALTY_CZK_KWH = 80.0
 PRE_NEG_BATT_EXPORT_MIN_SELL_CZK_KWH = 1.0
-PLANNER_BUILD_TAG = "2026-05-28-evening-export-soc-balance-v39"
+PLANNER_BUILD_TAG = "2026-05-29-neg-prep-observed-soc-v40"
 # Po t_detach v prep: necpát PV do bat (měkké; tvrdý hold přes soc_target z rampy).
 NEG_SELL_POST_DETACH_BCPV_DISCOURAGE_CZK_KWH = 250.0
 # Večer před neg dnem: výboj do sítě (měkký shortfall na ge_bat).
@@ -1077,34 +1077,35 @@ def _pre_neg_pv_export_forecast_cushion_ok_for_day(
     slots: list[PlanningSlot],
     battery: Any,
     first_neg_t: int,
-    soc_at_day_start_wh: float,
+    observed_soc_wh: float,
     *,
     neg_sell_phases_en: bool,
     soc_target_by_t: list[Optional[float]] | None = None,
 ) -> bool:
     """
-  Cushion pro jeden pražský den: usable A+B v sell<0 okně pokryje dobítí na soc_need[first_neg].
+    Cushion pro jeden pražský den: usable A+B v sell<0 okně pokryje dobítí na soc_need[first_neg].
+    Vstup SoC = pozorovaná telemetrie (ne trajektorie z předchozího solve).
     """
     if first_neg_t < 0 or first_neg_t >= len(slots):
         return False
     if neg_sell_phases_en and soc_target_by_t is not None:
         tgt = soc_target_by_t[first_neg_t]
         target_wh = float(tgt) if tgt is not None else float(battery.soc_max_wh)
-        usable_wh = _neg_sell_day_pv_usable_wh(
-            slots,
-            first_neg_t,
-            max_charge_power_w=float(battery.max_charge_power_w),
-            charge_efficiency=float(battery.charge_efficiency),
-        )
     else:
         target_wh = float(battery.soc_max_wh)
-        usable_wh = _neg_sell_day_pv_usable_wh(
-            slots,
-            first_neg_t,
-            max_charge_power_w=float(battery.max_charge_power_w),
-            charge_efficiency=float(battery.charge_efficiency),
-        )
-    needed_wh = max(0.0, target_wh - float(soc_at_day_start_wh))
+    soc_obs = max(
+        float(battery.min_soc_wh),
+        min(float(observed_soc_wh), float(battery.soc_max_wh)),
+    )
+    if soc_obs >= target_wh - 1e-3:
+        return True
+    usable_wh = _neg_sell_day_pv_usable_wh(
+        slots,
+        first_neg_t,
+        max_charge_power_w=float(battery.max_charge_power_w),
+        charge_efficiency=float(battery.charge_efficiency),
+    )
+    needed_wh = max(0.0, target_wh - soc_obs)
     if needed_wh < PRE_NEG_PV_EXPORT_MIN_NEEDED_WH:
         return True
     return usable_wh >= needed_wh * PRE_NEG_PV_EXPORT_FORECAST_MARGIN
@@ -1113,13 +1114,13 @@ def _pre_neg_pv_export_forecast_cushion_ok_for_day(
 def _pre_neg_pv_export_forecast_cushion_ok(
     slots: list[PlanningSlot],
     battery: Any,
-    current_soc_wh: float,
+    observed_soc_wh: float,
     first_neg_sell_idx: int | None,
     *,
     neg_sell_phases_en: bool,
     soc_target_by_t: list[Optional[float]] | None = None,
 ) -> bool:
-    """Zpětná kompatibilita: cushion pro první sell<0 v horizontu."""
+    """Zpětná kompatibilita: cushion pro první sell<0 v horizontu (pozorované SoC)."""
     if first_neg_sell_idx is None or first_neg_sell_idx <= 0:
         return False
     targets = soc_target_by_t
@@ -1129,7 +1130,7 @@ def _pre_neg_pv_export_forecast_cushion_ok(
         slots,
         battery,
         first_neg_sell_idx,
-        current_soc_wh,
+        observed_soc_wh,
         neg_sell_phases_en=neg_sell_phases_en,
         soc_target_by_t=targets,
     )
@@ -1161,7 +1162,7 @@ def _pre_neg_pv_export_slot_indices_for_day(
 def _pre_neg_pv_export_bundle(
     slots: list[PlanningSlot],
     battery: Any,
-    current_soc_wh: float,
+    observed_soc_wh: float,
     first_neg_buy_idx: int | None,
     *,
     neg_sell_phases_en: bool,
@@ -1169,11 +1170,11 @@ def _pre_neg_pv_export_bundle(
 ) -> tuple[set[int], dict[str, bool]]:
     """
     v36: pre-neg export per pražský den s vlastním cushion (A+B v neg okně dne).
+    v40: cushion vždy z pozorovaného SoC (telemetrie), bez řetězení modelových cílů mezi dny.
     """
     by_day = _neg_sell_indices_by_prague_day(slots)
     export_ts: set[int] = set()
     cushion_by_day: dict[str, bool] = {}
-    soc_est = float(current_soc_wh)
     for day in sorted(by_day.keys()):
         indices = by_day[day]
         if not indices:
@@ -1183,7 +1184,7 @@ def _pre_neg_pv_export_bundle(
             slots,
             battery,
             first_t,
-            soc_est,
+            observed_soc_wh,
             neg_sell_phases_en=neg_sell_phases_en,
             soc_target_by_t=soc_target_by_t,
         )
@@ -1194,12 +1195,6 @@ def _pre_neg_pv_export_bundle(
                 first_t,
                 first_neg_buy_idx,
             )
-        tgt0 = (
-            float(soc_target_by_t[first_t])
-            if soc_target_by_t and soc_target_by_t[first_t] is not None
-            else float(battery.soc_max_wh)
-        )
-        soc_est = max(float(battery.min_soc_wh), min(float(battery.soc_max_wh), tgt0))
     return export_ts, cushion_by_day
 
 
@@ -1268,6 +1263,64 @@ def _evening_discharge_before_neg_day_ts(
     return out
 
 
+def _night_baseload_buffer_wh_from_slots(
+    slots: list[PlanningSlot],
+    battery: Any,
+) -> float:
+    """Buffer Wh nad reserve pro noc (R__063 nebo % z asset_battery)."""
+    if not slots:
+        return 0.0
+    slot0 = slots[0]
+    buf = getattr(slot0, "night_baseload_buffer_wh", None)
+    if buf is not None:
+        return max(0.0, float(buf))
+    target = getattr(slot0, "night_baseload_target_wh", None)
+    if target is not None:
+        pct = float(getattr(battery, "planner_night_baseload_buffer_percent", 20.0) or 20.0)
+        return max(0.0, float(target) * pct / 100.0)
+    return 0.0
+
+
+def _neg_evening_discharge_budget_wh(
+    *,
+    observed_soc_wh: float,
+    reserve_soc_wh: float,
+    night_baseload_buffer_wh: float,
+) -> float:
+    """Wh k výboji nad reserve + noční buffer — z telemetrie, ne z LP trajektorie."""
+    return max(
+        0.0,
+        float(observed_soc_wh) - float(reserve_soc_wh) - float(night_baseload_buffer_wh),
+    )
+
+
+def _neg_evening_before_neg_push_indices(
+    slots: list[PlanningSlot],
+    candidate_ts: set[int],
+    *,
+    export_budget_wh: float,
+    per_slot_discharge_wh: float,
+) -> set[int]:
+    """Nejdražší kladné-sell sloty v kandidátech, dokud budget z pozorovaného SoC."""
+    if export_budget_wh < per_slot_discharge_wh * 0.5 or not candidate_ts:
+        return set()
+    ranked = sorted(
+        candidate_ts,
+        key=lambda t: (float(slots[t].sell_price), -t),
+        reverse=True,
+    )
+    out: set[int] = set()
+    cum_wh = 0.0
+    for t in ranked:
+        if float(slots[t].sell_price) < 0.0:
+            continue
+        if cum_wh + per_slot_discharge_wh > export_budget_wh + 1e-6:
+            break
+        out.add(t)
+        cum_wh += per_slot_discharge_wh
+    return out
+
+
 def _neg_evening_reserve_soc_anchors(
     slots: list[PlanningSlot],
     neg_sell_day_meta: dict[str, Any],
@@ -1987,6 +2040,10 @@ def solve_dispatch(
             else float(sm)
             for i, sm in enumerate(soc_min_series)
         ]
+    observed_soc_wh = max(
+        float(battery.min_soc_wh),
+        min(float(current_soc_wh), float(battery.soc_max_wh)),
+    )
     soc_headroom_applied_wh: float | None = None
     current_soc_wh, soc_headroom_applied_wh = _planner_soc_for_solver(
         current_soc_wh, battery
@@ -2196,12 +2253,14 @@ def solve_dispatch(
     pre_neg_cushion_by_day: dict[str, bool] = {}
     pre_neg_pv_export_ts: set[int] = set()
     neg_evening_before_neg_ts: set[int] = set()
+    neg_evening_push_ts: set[int] = set()
+    neg_evening_export_budget_wh: float | None = None
     neg_evening_reserve_anchors: list[tuple[int, float]] = []
     if om == "AUTO" and not purchase_fixed_pre and neg_sell_phases_en:
         pre_neg_pv_export_ts, pre_neg_cushion_by_day = _pre_neg_pv_export_bundle(
             slots,
             battery,
-            current_soc_wh,
+            observed_soc_wh,
             first_neg_buy_idx,
             neg_sell_phases_en=True,
             soc_target_by_t=neg_sell_soc_target_by_t,
@@ -2219,6 +2278,27 @@ def solve_dispatch(
             neg_sell_day_meta,
             battery,
         )
+        reserve_wh = float(
+            getattr(battery, "reserve_soc_wh", getattr(battery, "min_soc_wh", 0.0))
+        )
+        night_buf_wh = _night_baseload_buffer_wh_from_slots(slots, battery)
+        neg_evening_export_budget_wh = _neg_evening_discharge_budget_wh(
+            observed_soc_wh=observed_soc_wh,
+            reserve_soc_wh=reserve_wh,
+            night_baseload_buffer_wh=night_buf_wh,
+        )
+        per_slot_neg_eve_wh = max(
+            float(battery.max_discharge_power_w)
+            * float(battery.discharge_efficiency)
+            * INTERVAL_H,
+            0.0,
+        )
+        neg_evening_push_ts = _neg_evening_before_neg_push_indices(
+            slots,
+            neg_evening_before_neg_ts,
+            export_budget_wh=float(neg_evening_export_budget_wh),
+            per_slot_discharge_wh=per_slot_neg_eve_wh,
+        )
     elif om == "AUTO" and not purchase_fixed_pre:
         legacy_ok = bool(
             first_neg_sell_idx is not None
@@ -2226,7 +2306,7 @@ def solve_dispatch(
             and _pre_neg_pv_export_forecast_cushion_ok(
                 slots,
                 battery,
-                current_soc_wh,
+                observed_soc_wh,
                 first_neg_sell_idx,
                 neg_sell_phases_en=False,
             )
@@ -2528,7 +2608,7 @@ def solve_dispatch(
             sf_pe = pulp.LpVariable(f"pre_neg_pv_export_sf_{t_pe}", 0, cap_pe)
             pre_neg_pv_export_shortfall.append((t_pe, sf_pe, cap_pe))
         export_cap_evening = _battery_export_cap_w(battery, grid)
-        for t_ev in sorted(neg_evening_before_neg_ts):
+        for t_ev in sorted(neg_evening_push_ts):
             if t_ev not in discharge_export_slots:
                 continue
             sf_ev = pulp.LpVariable(
@@ -3267,7 +3347,7 @@ def solve_dispatch(
                 and t in discharge_export_slots
                 and (
                     t in evening_peak_export_ts
-                    or t in neg_evening_before_neg_ts
+                    or t in neg_evening_push_ts
                 )
             ):
                 export_soc_floor_t = float(min_soc_wh)
@@ -3745,7 +3825,7 @@ def solve_dispatch(
                     t in pre_neg_pv_export_ts if neg_sell_phases_en else None
                 ),
                 "neg_evening_before_neg": (
-                    t in neg_evening_before_neg_ts if neg_sell_phases_en else None
+                    t in neg_evening_push_ts if neg_sell_phases_en else None
                 ),
                 "neg_evening_reserve_anchor": (
                     any(t == ta for ta, _ in neg_evening_reserve_anchors)
@@ -3832,6 +3912,7 @@ def solve_dispatch(
         "planner_build_tag": PLANNER_BUILD_TAG,
         "inputs": {
             "current_soc_wh": float(current_soc_wh),
+            "observed_soc_wh": float(observed_soc_wh),
             "soc_headroom_applied_wh": soc_headroom_applied_wh,
             "operating_mode": operating_mode,
             "planner_version": planner_version_resolved,
@@ -3878,6 +3959,15 @@ def solve_dispatch(
                 slots[i].interval_start.isoformat()
                 for i in sorted(neg_evening_before_neg_ts)
             ],
+            "neg_evening_push_slots": [
+                slots[i].interval_start.isoformat()
+                for i in sorted(neg_evening_push_ts)
+            ],
+            "neg_evening_export_budget_wh": (
+                float(neg_evening_export_budget_wh)
+                if neg_evening_export_budget_wh is not None
+                else None
+            ),
             "neg_evening_reserve_soc_anchors": [
                 {
                     "slot": slots[t_a].interval_start.isoformat(),

backend/tests/test_planning_dispatch_milp.py

@@ -19,8 +19,11 @@ from services.planning_engine import (
     _in_night_battery_export_window,
     _neg_sell_day_phases,
     _neg_sell_phases_enabled,
+    _neg_evening_before_neg_push_indices,
+    _neg_evening_discharge_budget_wh,
     _neg_evening_reserve_soc_anchors,
     _pre_neg_pv_export_bundle,
+    _pre_neg_pv_export_forecast_cushion_ok_for_day,
     _prague_calendar_date,
     _pre_neg_buy_soc_ceiling_wh,
     _pre_neg_peak_sell_idx,
@@ -4472,6 +4475,156 @@ class NegSellPrepWindowV36Tests(unittest.TestCase):
         self.assertLessEqual(soc_wh, cap_wh + 800.0)
         eve_slots = snap["inputs"].get("neg_evening_before_neg_slots") or []
         self.assertGreater(len(eve_slots), 8)
+        push_slots = snap["inputs"].get("neg_evening_push_slots") or []
+        self.assertGreater(len(push_slots), 0)
+        self.assertIn("observed_soc_wh", snap["inputs"])
+        self.assertIsNotNone(snap["inputs"].get("neg_evening_export_budget_wh"))
+
+
+class ObservedSocNegPrepTests(unittest.TestCase):
+    """v40: neg-prep a večerní výboj z pozorovaného SoC (telemetrie), ne z LP trajektorie."""
+
+    def test_cushion_ok_when_observed_above_prep_target(self) -> None:
+        base = datetime(2026, 6, 11, 6, 0, tzinfo=ZoneInfo("Europe/Prague")).astimezone(
+            timezone.utc
+        )
+        slots: list[PlanningSlot] = []
+        for i in range(48):
+            local = (base + timedelta(minutes=15 * i)).astimezone(
+                ZoneInfo("Europe/Prague")
+            )
+            sell = -0.2 if local.hour >= 10 else 3.0
+            slots.append(
+                PlanningSlot(
+                    interval_start=base + timedelta(minutes=15 * i),
+                    buy_price=2.0,
+                    sell_price=sell,
+                    pv_a_forecast_w=5000,
+                    pv_b_forecast_w=8000,
+                    load_baseline_w=500,
+                    ev1_connected=False,
+                    ev2_connected=False,
+                    allow_charge=True,
+                    allow_discharge_export=True,
+                )
+            )
+        bat = _battery(uc_wh=64_000.0, max_pct=95.0)
+        bat.planner_neg_sell_prep_soc_percent = 80.0
+        bat.planner_neg_sell_full_soc_tail_slots = 4
+        _ph, tg, _w, _meta = _neg_sell_day_phases(slots, bat)
+        first_neg = next(i for i, s in enumerate(slots) if float(s.sell_price) < 0)
+        tgt = float(tg[first_neg] or bat.soc_max_wh)
+        observed_high = tgt + 5000.0
+        self.assertTrue(
+            _pre_neg_pv_export_forecast_cushion_ok_for_day(
+                slots,
+                bat,
+                first_neg,
+                observed_high,
+                neg_sell_phases_en=True,
+                soc_target_by_t=tg,
+            ),
+            "pozorované SoC nad prep cílem → cushion bez forecastu",
+        )
+
+    def test_pre_neg_bundle_uses_observed_not_model_chain(self) -> None:
+        base = datetime(2026, 6, 10, 10, 0, tzinfo=ZoneInfo("Europe/Prague")).astimezone(
+            timezone.utc
+        )
+        slots: list[PlanningSlot] = []
+        for i in range(120):
+            local = (base + timedelta(minutes=15 * i)).astimezone(
+                ZoneInfo("Europe/Prague")
+            )
+            h = local.hour + local.minute / 60.0
+            if local.date().day == 10:
+                sell = -0.2 if h >= 14 else 2.5
+            elif local.date().day == 11:
+                sell = -0.2 if 9 <= h < 15 else 2.8
+            else:
+                sell = 2.5
+            slots.append(
+                PlanningSlot(
+                    interval_start=base + timedelta(minutes=15 * i),
+                    buy_price=2.0,
+                    sell_price=sell,
+                    pv_a_forecast_w=7000,
+                    pv_b_forecast_w=9000,
+                    load_baseline_w=500,
+                    ev1_connected=False,
+                    ev2_connected=False,
+                    allow_charge=True,
+                    allow_discharge_export=True,
+                )
+            )
+        bat = _battery(uc_wh=64_000.0, max_pct=95.0)
+        bat.planner_neg_sell_prep_soc_percent = 80.0
+        bat.planner_neg_sell_full_soc_tail_slots = 4
+        _ph, tg, _w, meta = _neg_sell_day_phases(slots, bat)
+        observed = 0.72 * bat.soc_max_wh
+        export_ts, cushion = _pre_neg_pv_export_bundle(
+            slots,
+            bat,
+            observed,
+            None,
+            neg_sell_phases_en=True,
+            soc_target_by_t=tg,
+        )
+        self.assertTrue(all(cushion.values()))
+        if len(meta.get("days", [])) >= 2:
+            second_first = int(meta["days"][1]["first_neg_idx"])
+            second_morning = [
+                t for t in export_ts if t < second_first and float(slots[t].sell_price) >= 0.0
+            ]
+            self.assertGreater(len(second_morning), 0)
+
+    def test_neg_evening_push_scales_with_observed_soc(self) -> None:
+        base = datetime(2026, 6, 10, 18, 0, tzinfo=ZoneInfo("Europe/Prague")).astimezone(
+            timezone.utc
+        )
+        slots: list[PlanningSlot] = []
+        for i in range(16):
+            slots.append(
+                PlanningSlot(
+                    interval_start=base + timedelta(minutes=15 * i),
+                    buy_price=2.0,
+                    sell_price=3.0 + i * 0.1,
+                    pv_a_forecast_w=0,
+                    pv_b_forecast_w=0,
+                    load_baseline_w=500,
+                    ev1_connected=False,
+                    ev2_connected=False,
+                    allow_charge=True,
+                    allow_discharge_export=True,
+                )
+            )
+        candidates = set(range(16))
+        bat = _battery(uc_wh=64_000.0, max_pct=95.0, arb_pct=20.0)
+        reserve = float(bat.reserve_soc_wh)
+        per_slot = 3000.0
+        budget_low = _neg_evening_discharge_budget_wh(
+            observed_soc_wh=reserve + 2000.0,
+            reserve_soc_wh=reserve,
+            night_baseload_buffer_wh=1000.0,
+        )
+        budget_high = _neg_evening_discharge_budget_wh(
+            observed_soc_wh=0.70 * bat.soc_max_wh,
+            reserve_soc_wh=reserve,
+            night_baseload_buffer_wh=1000.0,
+        )
+        push_low = _neg_evening_before_neg_push_indices(
+            slots,
+            candidates,
+            export_budget_wh=budget_low,
+            per_slot_discharge_wh=per_slot,
+        )
+        push_high = _neg_evening_before_neg_push_indices(
+            slots,
+            candidates,
+            export_budget_wh=budget_high,
+            per_slot_discharge_wh=per_slot,
+        )
+        self.assertLess(len(push_low), len(push_high))
 
 
 class SocBalanceDischargeTests(unittest.TestCase):