Fáze 3.2: solver_v2 — čisté ekonomické jádro plánovače

services/planning/solver_v2.py: MILP s objective = reálné peníze (cash +
degradace − terminal SoC value z DB faktoru). Tvrdá pravidla: bilance, SoC
dynamika, breaker (tvrdý), curtail jen A, GEN cutoff binárka, neg-buy/neg-sell
export bloky, export z baterie ⇒ arb floor (p.19), zákaz současného imp+exp,
EV deadline (placený slack 50 Kč/kWh místo infeasibility), TUV look-ahead,
provozní režimy. SQL masky allow_* vědomě ignorovány (heuristika, ne fyzika).

solver_v2_eval.py: v2 vs v1 na golden fixtures (SoC-fér):
  v2 lepší na VŠECH 5 řešitelných (+231.5 Kč ≈ +22 %), extreme_neg_buy den
  v1=INFEASIBLE → v2 OK (−674.5 Kč). Časy 0.4–10 s (2× na time limitu — TODO).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Dusan Vojacek
2026-06-11 14:19:32 +02:00
parent 368291e562
commit 90a85b2727
2 changed files with 500 additions and 0 deletions

View File

@@ -0,0 +1,100 @@
#!/usr/bin/env python3
"""
Fáze 3 vyhodnocení solver_v2 (čisté jádro) proti v1 na golden fixtures.
Replay STEJNOU cestou jako golden gate (_load_site_context + _load_slots nad
FixtureDB), ale přes services.planning.solver_v2.solve_dispatch_v2. Porovnání
s golden snapshoty v1 (SoC-fér: koncový SoC obou oceněn terminal cenou v2).
Spouštět z backend/: python3 ../scripts/harness/solver_v2_eval.py
"""
from __future__ import annotations
import asyncio
import importlib.util
import json
import sys
from datetime import datetime
from pathlib import Path
BACKEND = Path(__file__).resolve().parents[2] / "backend"
sys.path.insert(0, str(BACKEND))
from services import planning_engine as pe # noqa: E402
from services.planning import solver_v2 as v2 # noqa: E402
_spec = importlib.util.spec_from_file_location(
"golden_replay", BACKEND / "tests" / "test_golden_replay.py"
)
_golden = importlib.util.module_from_spec(_spec)
sys.modules["golden_replay"] = _golden
_spec.loader.exec_module(_golden)
FIXTURES = sorted((BACKEND / "tests" / "golden" / "fixtures").glob("*.json"))
SNAPSHOTS = BACKEND / "tests" / "golden" / "snapshots"
def _replay_v2(fixture: dict):
async def _run():
db = _golden._FixtureDB(fixture)
meta = fixture["meta"]
(battery, heat_pump, grid, vehicles, ev_sessions, soc_wh, tuv_temp,
operating_mode, tuv_stats) = await pe._load_site_context(int(meta["site_id"]), db)
slots = await pe._load_slots(
int(meta["site_id"]),
datetime.fromisoformat(meta["window_from"]),
datetime.fromisoformat(meta["window_to"]),
db,
soc_wh=soc_wh,
)
results, ms, snap = v2.solve_dispatch_v2(
slots, battery, heat_pump, grid, ev_sessions, vehicles,
soc_wh, tuv_temp,
tuv_delta_stats=tuv_stats,
operating_mode=operating_mode or "AUTO",
)
return results, ms, snap, battery
return asyncio.run(_run())
def main() -> None:
header = f"{'fixture':<42} {'v1':>9} {'v2':>9} {'Δ':>8} {'v2 ms':>6} pozn."
print("# solver_v2 vs v1 — modelovaný cashflow, SoC-fér (Kč/horizont; Δ<0 = v2 lepší)")
print()
print(header)
print("-" * len(header))
tot1 = tot2 = 0.0
solved_both = 0
for path in FIXTURES:
fixture = json.loads(path.read_text(encoding="utf-8"))
try:
results, ms, snap, battery = _replay_v2(fixture)
except Exception as exc:
print(f"{path.stem:<42} {'?':>9} {'CHYBA':>9} {'':>8} {exc}")
continue
usable = float(battery.usable_capacity_wh)
term = float(snap["inputs"]["terminal_czk_per_wh"])
v2_cash = sum(r.cashflow_czk for r in results)
v2_soc_end = results[-1].battery_soc_target / 100.0 * usable
v2_adj = v2_cash - v2_soc_end * term
snap1 = json.loads((SNAPSHOTS / path.name).read_text(encoding="utf-8"))
if "solver_error" in snap1:
print(f"{path.stem:<42} {'INFEAS':>9} {v2_adj:>9.1f} {'':>8} {ms:>6} v1 selhal, v2 OK")
continue
v1_cash = snap1["totals"]["cashflow_czk"]
v1_soc_end = snap1["slots"][-1]["battery_soc_target"] / 100.0 * usable
v1_adj = v1_cash - v1_soc_end * term
d = v2_adj - v1_adj
tot1 += v1_adj
tot2 += v2_adj
solved_both += 1
print(f"{path.stem:<42} {v1_adj:>9.1f} {v2_adj:>9.1f} {d:>8.1f} {ms:>6}")
print("-" * len(header))
if solved_both:
print(f"{'CELKEM (oba řešitelné)':<42} {tot1:>9.1f} {tot2:>9.1f} {tot2 - tot1:>8.1f}")
if __name__ == "__main__":
main()