Fáze 0: ekonomický regresní harness plánovače
- scripts/harness/extract_fixtures.py: extrakce vstupů solveru (fn_planning_site_context + fn_load_planning_slots_full) do JSON fixtures - backend/tests/test_golden_replay.py: golden gate — replay fixtures přes solve_dispatch_two_pass, bit-perfektní diff proti snapshotům (GOLDEN_UPDATE=1 pro vědomou regeneraci); 4 scénáře: home-01 neg-sell extrém / normal, BA81, KV1 - scripts/harness/economics_report.py: actual (audit_interval) vs oracle MILP (perfect hindsight, čistá ekonomika bez heuristických penalt), SoC-adjusted Baseline home-01 2026-05-12..06-09: GAP 2185 Kč / 29 dní (~27 %). Známý stav: 4/124 testů test_planning_dispatch_milp.py failuje už na main. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Dusan Vojacek
193
backend/tests/test_golden_replay.py
Normal file
193
backend/tests/test_golden_replay.py
Normal file
@@ -0,0 +1,193 @@
|
||||
"""
|
||||
Fáze 0 – golden replay gate plánovače (bez DB).
|
||||
|
||||
Pro každou fixture v tests/golden/fixtures/ (kompletní vstupy solveru zmrazené
|
||||
z reálné DB skriptem scripts/harness/extract_fixtures.py) spustí
|
||||
solve_dispatch_two_pass a porovná normalizovaný výstup s golden snapshotem
|
||||
v tests/golden/snapshots/.
|
||||
|
||||
Účel: regresní brána pro dekompozici planning_engine.py — identity refactor
|
||||
musí držet výstupy bit-perfektně (floaty zaokrouhleny na 4 d.m.).
|
||||
|
||||
Regenerace snapshotů (vědomá změna chování):
|
||||
GOLDEN_UPDATE=1 python3 -m pytest tests/test_golden_replay.py -q
|
||||
|
||||
Replay jde STEJNOU cestou jako produkce: _load_site_context + _load_slots nad
|
||||
fixture stubem DB → žádná duplikace mapování DB → objekty.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import os
|
||||
import unittest
|
||||
from datetime import datetime
|
||||
from pathlib import Path
|
||||
|
||||
from services import planning_engine as pe
|
||||
|
||||
GOLDEN_DIR = Path(__file__).resolve().parent / "golden"
|
||||
FIXTURES_DIR = GOLDEN_DIR / "fixtures"
|
||||
SNAPSHOTS_DIR = GOLDEN_DIR / "snapshots"
|
||||
|
||||
_DT_SLOT_KEYS = ("interval_start", "charge_acquisition_cutoff_at")
|
||||
|
||||
|
||||
class _FixtureDB:
|
||||
"""Stub asyncpg connection: vrací zmrazený context a sloty z fixture."""
|
||||
|
||||
def __init__(self, fixture: dict):
|
||||
self._fixture = fixture
|
||||
|
||||
async def fetchval(self, query: str, *args):
|
||||
assert "fn_planning_site_context" in query, f"Nečekaný fetchval: {query!r}"
|
||||
return json.dumps(self._fixture["context_json"])
|
||||
|
||||
async def fetch(self, query: str, *args):
|
||||
assert "fn_load_planning_slots_full" in query, f"Nečekaný fetch: {query!r}"
|
||||
rows: list[dict] = []
|
||||
for raw in self._fixture["slot_rows"]:
|
||||
d = dict(raw)
|
||||
for key in _DT_SLOT_KEYS:
|
||||
if d.get(key):
|
||||
d[key] = datetime.fromisoformat(d[key])
|
||||
rows.append(d)
|
||||
return rows
|
||||
|
||||
|
||||
def _round(val: float, places: int = 4) -> float:
|
||||
out = round(float(val), places)
|
||||
return 0.0 if out == 0.0 else out # normalizace -0.0
|
||||
|
||||
|
||||
def _normalize_results(results: list) -> dict:
|
||||
rows = []
|
||||
for r in results:
|
||||
rows.append(
|
||||
{
|
||||
"interval_start": r.interval_start.isoformat(),
|
||||
"battery_setpoint_w": int(r.battery_setpoint_w),
|
||||
"battery_soc_target": _round(r.battery_soc_target, 2),
|
||||
"grid_setpoint_w": int(r.grid_setpoint_w),
|
||||
"export_limit_w": int(r.export_limit_w),
|
||||
"export_mode": r.export_mode,
|
||||
"deye_physical_mode": r.deye_physical_mode,
|
||||
"deye_gen_cutoff_enabled": r.deye_gen_cutoff_enabled,
|
||||
"ev1_setpoint_w": r.ev1_setpoint_w,
|
||||
"ev2_setpoint_w": r.ev2_setpoint_w,
|
||||
"ev1_via_bat_w": int(r.ev1_via_bat_w),
|
||||
"ev2_via_bat_w": int(r.ev2_via_bat_w),
|
||||
"heat_pump_enabled": bool(r.heat_pump_enabled),
|
||||
"heat_pump_setpoint_w": int(r.heat_pump_setpoint_w),
|
||||
"pv_a_curtailed_w": int(r.pv_a_curtailed_w),
|
||||
"expected_cost_czk": _round(r.expected_cost_czk),
|
||||
"cashflow_czk": _round(r.cashflow_czk),
|
||||
"battery_arbitrage_czk": _round(r.battery_arbitrage_czk),
|
||||
"penalty_czk": _round(r.penalty_czk),
|
||||
"green_bonus_czk": _round(r.green_bonus_czk),
|
||||
}
|
||||
)
|
||||
totals = {
|
||||
"slots": len(rows),
|
||||
"expected_cost_czk": _round(sum(r["expected_cost_czk"] for r in rows), 3),
|
||||
"cashflow_czk": _round(sum(r["cashflow_czk"] for r in rows), 3),
|
||||
"penalty_czk": _round(sum(r["penalty_czk"] for r in rows), 3),
|
||||
"grid_import_slots": sum(1 for r in rows if r["grid_setpoint_w"] > 0),
|
||||
"grid_export_slots": sum(1 for r in rows if r["grid_setpoint_w"] < 0),
|
||||
"curtail_slots": sum(1 for r in rows if r["pv_a_curtailed_w"] > 0),
|
||||
}
|
||||
return {"totals": totals, "slots": rows}
|
||||
|
||||
|
||||
def _replay_fixture(fixture: dict) -> dict:
|
||||
async def _run() -> dict:
|
||||
db = _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 = pe.solve_dispatch_two_pass(
|
||||
slots,
|
||||
battery,
|
||||
heat_pump,
|
||||
grid,
|
||||
ev_sessions,
|
||||
vehicles,
|
||||
soc_wh,
|
||||
tuv_temp,
|
||||
tuv_delta_stats=tuv_stats,
|
||||
operating_mode=operating_mode or "AUTO",
|
||||
planner_version=pe._planner_engine_version(),
|
||||
)
|
||||
return _normalize_results(results)
|
||||
|
||||
return asyncio.run(_run())
|
||||
|
||||
|
||||
def _fixture_paths() -> list[Path]:
|
||||
return sorted(FIXTURES_DIR.glob("*.json"))
|
||||
|
||||
|
||||
class GoldenReplayTests(unittest.TestCase):
|
||||
maxDiff = None
|
||||
|
||||
def test_fixtures_exist(self) -> None:
|
||||
self.assertTrue(
|
||||
_fixture_paths(),
|
||||
f"Žádné fixtures v {FIXTURES_DIR} – spusť scripts/harness/extract_fixtures.py",
|
||||
)
|
||||
|
||||
|
||||
def _make_test(path: Path):
|
||||
def test(self: GoldenReplayTests) -> None:
|
||||
fixture = json.loads(path.read_text(encoding="utf-8"))
|
||||
actual = _replay_fixture(fixture)
|
||||
snap_path = SNAPSHOTS_DIR / path.name
|
||||
if os.environ.get("GOLDEN_UPDATE") == "1":
|
||||
SNAPSHOTS_DIR.mkdir(parents=True, exist_ok=True)
|
||||
snap_path.write_text(
|
||||
json.dumps(actual, ensure_ascii=False, indent=1) + "\n", encoding="utf-8"
|
||||
)
|
||||
return
|
||||
self.assertTrue(
|
||||
snap_path.exists(),
|
||||
f"Chybí snapshot {snap_path.name} – vygeneruj přes GOLDEN_UPDATE=1",
|
||||
)
|
||||
expected = json.loads(snap_path.read_text(encoding="utf-8"))
|
||||
self.assertEqual(
|
||||
expected["totals"],
|
||||
actual["totals"],
|
||||
f"{path.name}: změna agregátů plánu (totals)",
|
||||
)
|
||||
self.assertEqual(
|
||||
expected["slots"],
|
||||
actual["slots"],
|
||||
f"{path.name}: změna plánu per slot",
|
||||
)
|
||||
|
||||
return test
|
||||
|
||||
|
||||
for _path in _fixture_paths():
|
||||
_name = "test_golden_" + _path.stem.replace("-", "_").replace(".", "_")
|
||||
setattr(GoldenReplayTests, _name, _make_test(_path))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main()
|
||||
Reference in New Issue
Block a user