ems/scripts/harness/penalty_audit.py

#!/usr/bin/env python3
"""
Fáze 2 – penalty audit: změř přínos každé heuristické penalty v objective.

Pro každou ekonomickou konstantu (penalty/reward/bonus/surcharge) ji vynuluje
ve VŠECH modulech, kde je importovaná (fasáda: planning_engine, planning.heuristics,
planning.constants), přehraje golden fixtures (tests/golden/fixtures/) přes
solve_dispatch_two_pass a porovná s baseline:

  - Δcashflow  : změna reálných peněz plánu (− = plán vydělá víc bez penalty),
  - Δslotů     : kolik slotů změnilo battery/grid setpoint (chování),
  - bind       : jestli penalta vůbec něco dělá (Δ=0 na všech fixtures = mrtvá).

POZOR na interpretaci: penalty mění modelované peníze za robustnost vůči chybě
forecastu. Záporná Δcashflow ⇒ penalta v modelu stojí peníze — to samo o sobě
neznamená „smazat“; mrtvé penalty (bind=NE) ale smazat lze bezpečně.
Druhý krok auditu = economics_report nad reálnými dny.

Spouštět z backend/:  python3 ../scripts/harness/penalty_audit.py [--only NÁZEV]
"""

from __future__ import annotations

import argparse
import importlib.util
import json
import re
import sys
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 constants as C  # noqa: E402
from services.planning import heuristics as H  # noqa: E402

# replay funkce z golden testu (bez duplikace logiky)
_spec = importlib.util.spec_from_file_location(
    "golden_replay", BACKEND / "tests" / "test_golden_replay.py"
)
_golden = importlib.util.module_from_spec(_spec)
_spec.loader.exec_module(_golden)

FIXTURES = sorted((BACKEND / "tests" / "golden" / "fixtures").glob("*.json"))

# Ekonomické konstanty k auditu: penalty/reward/bonus/surcharge/incentive/discourage
AUDIT_PATTERN = re.compile(
    r"(PENALTY|SHORTFALL|DISCOURAGE|REWARD|SURCHARGE|INCENTIVE|BONUS)", re.I
)
MODULES = (pe, C, H)


def _audit_names() -> list[str]:
    names = []
    for n in dir(C):
        if n.startswith("_") or not n.isupper():
            continue
        if AUDIT_PATTERN.search(n) and isinstance(getattr(C, n), (int, float)):
            names.append(n)
    return sorted(names)


def _set_const(name: str, value: float) -> dict:
    saved = {}
    for mod in MODULES:
        if hasattr(mod, name):
            saved[mod.__name__] = getattr(mod, name)
            setattr(mod, name, value)
    return saved


def _restore_const(name: str, saved: dict) -> None:
    for mod in MODULES:
        if mod.__name__ in saved:
            setattr(mod, name, saved[mod.__name__])


def _replay_all() -> dict[str, dict]:
    out = {}
    for path in FIXTURES:
        fixture = json.loads(path.read_text(encoding="utf-8"))
        out[path.stem] = _golden._replay_fixture(fixture)
    return out


def _diff(base: dict, new: dict) -> tuple[float, float, int, list[str]]:
    """(Δcashflow, Δpenalty, změněné sloty, změny feasibility) napříč fixtures."""
    d_cash = d_pen = 0.0
    changed = 0
    feas: list[str] = []
    for key, b in base.items():
        n = new[key]
        b_err = "solver_error" in b
        n_err = "solver_error" in n
        if b_err or n_err:
            if b_err and not n_err:
                feas.append(f"{key}: INFEASIBLE → OK!")
                changed += 1
            elif n_err and not b_err:
                feas.append(f"{key}: OK → INFEASIBLE")
                changed += 1
            continue
        d_cash += n["totals"]["cashflow_czk"] - b["totals"]["cashflow_czk"]
        d_pen += n["totals"]["penalty_czk"] - b["totals"]["penalty_czk"]
        for rb, rn in zip(b["slots"], n["slots"]):
            if (
                rb["battery_setpoint_w"] != rn["battery_setpoint_w"]
                or rb["grid_setpoint_w"] != rn["grid_setpoint_w"]
                or rb["pv_a_curtailed_w"] != rn["pv_a_curtailed_w"]
            ):
                changed += 1
    return d_cash, d_pen, changed, feas


def main() -> None:
    ap = argparse.ArgumentParser()
    ap.add_argument("--only", default=None, help="audit jen jedné konstanty")
    args = ap.parse_args()

    names = [args.only] if args.only else _audit_names()
    print(f"# Penalty audit — {len(names)} konstant × {len(FIXTURES)} fixtures")
    print("# Δcashflow: záporné = plán bez penalty vydělá víc (modelované Kč za horizonty fixtures)")
    print()

    baseline = _replay_all()
    base_cash = sum(r["totals"]["cashflow_czk"] for r in baseline.values() if "totals" in r)
    base_pen = sum(r["totals"]["penalty_czk"] for r in baseline.values() if "totals" in r)
    infeas = [k for k, r in baseline.items() if "solver_error" in r]
    print(f"baseline: cashflow {base_cash:.1f} Kč, penalty {base_pen:.1f} Kč; infeasible fixtures: {infeas}\n")

    header = f"{'konstanta':<55} {'hodnota':>9} {'Δcash':>8} {'Δpenalty':>9} {'Δsloty':>6}  bind"
    print(header)
    print("-" * len(header))

    rows = []
    for name in names:
        value = getattr(C, name)
        saved = _set_const(name, 0.0)
        try:
            result = _replay_all()
        except Exception as exc:  # infeasible apod. — informace sama o sobě
            print(f"{name:<55} {value:>9} {'ERROR':>8} {str(exc)[:40]}")
            _restore_const(name, saved)
            continue
        _restore_const(name, saved)
        d_cash, d_pen, changed, feas = _diff(baseline, result)
        bind = "NE (mrtvá?)" if changed == 0 and abs(d_cash) < 0.05 else "ano"
        if feas:
            bind = " | ".join(feas)
        rows.append((name, value, d_cash, d_pen, changed, bind))
        print(f"{name:<55} {value:>9} {d_cash:>8.1f} {d_pen:>9.1f} {changed:>6}  {bind}")

    dead = [r[0] for r in rows if r[5].startswith("NE")]
    print(f"\nMrtvé penalty (žádný vliv na 4 fixtures): {len(dead)}")
    for n in dead:
        print(f"  - {n}")


if __name__ == "__main__":
    main()