- 1
The agent decidesIt reads the district, prices and forecasts and proposes cooling, hot water and battery actions as strict JSON.
- - 2
The safety envelope checksPlain code enforces ranges, ramps, charge limits, comfort, the price spread and the peak cap, and logs every clamp.
+ - 2
The safety envelope checksPlain code enforces ranges, ramps, charge limits, comfort, the price spread, the peak cap and the cooling trim hold, and logs every clamp.
- 3
CityLearn simulatesOne hour of three buildings: indoor temperature, heat pumps, batteries, hot water and rooftop solar.
- 4
CityLearn scoresIts own KPI evaluation compares the month with the no-control reference: cost, carbon, comfort and the grid.
diff --git a/hyperbolic/__init__.py b/hyperbolic/__init__.py
index c94a41e..d1f8295 100644
--- a/hyperbolic/__init__.py
+++ b/hyperbolic/__init__.py
@@ -1,3 +1,3 @@
"""Hyperbolic: a supervisory controller for CityLearn district simulations."""
-__version__ = "0.7.0"
+__version__ = "0.8.0"
diff --git a/hyperbolic/cli.py b/hyperbolic/cli.py
index 9e3b987..66d7ccb 100644
--- a/hyperbolic/cli.py
+++ b/hyperbolic/cli.py
@@ -174,7 +174,8 @@ def _against(run_dir: Path, hours: int) -> tuple[str, dict, Path]:
raise ValueError(f"it covers {r.get('steps')} hours, the compared runs {hours}")
sup = m.get("supervisor") or {}
if sup.get("kind") == "scripted":
- label = "rule-based" if int(sup.get("prompt_version", 1)) <= RULE_PROMPT_VERSION else "v4 script"
+ pv = int(sup.get("prompt_version", 1))
+ label = "rule-based" if pv <= RULE_PROMPT_VERSION else f"v{pv} script"
else:
label = str(sup.get("model_key") or sup.get("kind"))
return label, r.get("district_kpis") or {}, run_dir
@@ -214,8 +215,9 @@ def build_parser() -> argparse.ArgumentParser:
help="with --scripted, send a bad reply now and then to exercise clamps and fallback")
r.add_argument("--prompt-version", type=int, default=None, choices=PROMPT_VERSIONS,
help=f"observation and prompt version (default: {PROMPT_VERSION}; with --scripted, "
- f"{RULE_PROMPT_VERSION}, the 0.4.0 rule-based controller, while version 4 "
- "runs the version 4 strategy literally)")
+ f"{RULE_PROMPT_VERSION}, the 0.4.0 rule-based controller, while versions 4 and 5 "
+ "run their strategies literally and version 6 runs the version 5 strategy with the "
+ "trim hold)")
r.set_defaults(func=cmd_run)
rp = sub.add_parser("replay", help="re-run a saved run from its files with zero model or simulator calls")
diff --git a/hyperbolic/envelope.py b/hyperbolic/envelope.py
index b5b88dd..b9a31fc 100644
--- a/hyperbolic/envelope.py
+++ b/hyperbolic/envelope.py
@@ -14,11 +14,17 @@
7. comfort: in occupied hours near the edge of CityLearn's comfort band the heat
pump may not cool less (too warm) or more (too cool) than the baseline action,
and with overcool_guard_unoccupied no extra cooling when already cool at night
+ 8. trim hold (0.8.0): a cooling trim, once on, is carried in code. While the room is
+ occupied and more than trim_hold_c below its setpoint it stays at least as deep as
+ last hour; within trim_hold_c it eases off by at most trim_release_step per hour; it
+ ends at once at or above the setpoint, in an unoccupied room, or when the baseline
+ has no cooling to trim. The model can end one early only with end_trim and a reason
Then, across the district:
- 8. peak limit: storage charging is trimmed so the district's load this hour stays at
+ 9. peak limit: storage charging is trimmed so the district's load this hour stays at
or below the peak target (signals.peak_target_kwh), battery first, then the tank
A null action, or any invalid reply, uses the baseline action for that device.
-Every clamp and fallback is returned with a reason so the runner can log it.
+Every clamp and fallback is returned with a reason so the runner can log it, and every
+trim hold action (start, hold, ease, release, end, early end) is recorded with its reason.
"""
from __future__ import annotations
@@ -44,12 +50,18 @@ class DeviceActions(BaseModel):
battery: float | None
+class DeviceActionsV6(DeviceActions):
+ """Prompt version 6 (0.8.0): a building may also end its held cooling trim early, with a reason."""
+
+ end_trim: str | None = Field(default=None, min_length=1, max_length=dc.REASON_MAX_CHARS)
+
+
@lru_cache(maxsize=8)
-def proposal_model(buildings: tuple[str, ...]) -> type[BaseModel]:
+def proposal_model(buildings: tuple[str, ...], end_trim: bool = False) -> type[BaseModel]:
"""The only shape the model is allowed to return: a reason plus one entry per building."""
fields: dict[str, Any] = {"reason": (str, Field(min_length=1, max_length=dc.REASON_MAX_CHARS))}
for b in buildings:
- fields[b] = (DeviceActions, ...)
+ fields[b] = (DeviceActionsV6 if end_trim else DeviceActions, ...)
return create_model(
"Proposal", __config__=ConfigDict(extra="forbid", strict=True, allow_inf_nan=False), **fields
)
@@ -59,7 +71,7 @@ def _nullable_number() -> dict:
return {"anyOf": [{"type": "number"}, {"type": "null"}]}
-def proposal_json_schema(buildings: tuple[str, ...]) -> dict:
+def proposal_json_schema(buildings: tuple[str, ...], end_trim: bool = False) -> dict:
"""Hand-written JSON schema for structured output. Mirrors proposal_model."""
dev = {
"type": "object",
@@ -67,14 +79,19 @@ def proposal_json_schema(buildings: tuple[str, ...]) -> dict:
"required": list(DEVICE_KEYS),
"additionalProperties": False,
}
+ if end_trim:
+ dev["properties"]["end_trim"] = {"anyOf": [{"type": "string"}, {"type": "null"}]}
+ dev["required"].append("end_trim")
props: dict[str, Any] = {"reason": {"type": "string"}}
for b in buildings:
props[b] = dev
return {"type": "object", "properties": props, "required": ["reason", *buildings], "additionalProperties": False}
-def parse_reply(text: str | None, buildings: tuple[str, ...]) -> tuple[dict | None, str | None]:
- """Return ({"reason", "buildings": {b: {device: value}}}, None) or (None, why it was rejected)."""
+def parse_reply(text: str | None, buildings: tuple[str, ...], end_trim: bool = False) -> tuple[dict | None, str | None]:
+ """Return ({"reason", "buildings": {b: {device: value}}}, None) or (None, why it was rejected).
+ With end_trim (prompt version 6) the result also has "end_trim": {b: reason} for every
+ building whose reply ends its held cooling trim early."""
if text is None or not text.strip():
return None, "empty or missing reply"
try:
@@ -85,12 +102,16 @@ def parse_reply(text: str | None, buildings: tuple[str, ...]) -> tuple[dict | No
return None, "reply is not a JSON object"
data = _ints_to_floats(data)
try:
- p = proposal_model(tuple(buildings)).model_validate(data)
+ p = proposal_model(tuple(buildings), end_trim).model_validate(data)
except ValidationError as exc:
first = exc.errors()[0]
loc = ".".join(str(x) for x in first["loc"])
return None, f"reply failed schema at {loc or 'root'}: {first['msg']} ({exc.error_count()} error(s))"
- return {"reason": p.reason, "buildings": {b: getattr(p, b).model_dump() for b in buildings}}, None
+ out = {"reason": p.reason,
+ "buildings": {b: {d: getattr(getattr(p, b), d) for d in DEVICE_KEYS} for b in buildings}}
+ if end_trim:
+ out["end_trim"] = {b: getattr(p, b).end_trim for b in buildings if getattr(p, b).end_trim is not None}
+ return out, None
def _ints_to_floats(obj: Any) -> Any:
@@ -133,6 +154,14 @@ class EnvelopeConfig:
# its round trip and standing losses and a cheaper hour is coming (signals.charge_spread).
# Physical, not fitted: the efficiency and loss come from the battery itself.
spread_check: bool = True
+ # Added in 0.8.0. Carry a cooling trim in code once it is on (Envelope._hold_trim). The
+ # numbers are the version 5 strategy's (StrategyParamsV5.trim_hold_c and trim_step); a trim
+ # shallower than trim_min_depth counts as off, so rounding never starts one, and a reply that
+ # misses the held depth by less than it is taken as rounding, not as a release.
+ trim_hold: bool = True
+ trim_hold_c: float = 0.5
+ trim_release_step: float = 0.02
+ trim_min_depth: float = 0.002
def as_dict(self) -> dict:
return dict(self.__dict__)
@@ -142,14 +171,20 @@ def from_manifest(cls, data: dict | None) -> "EnvelopeConfig":
"""Rebuild the config a run used. Fields a run did not record keep the behavior
they had before they existed, so older runs replay exactly."""
legacy = {"cooling_max_deviation": None, "overcool_guard_unoccupied": False, "peak_limit": False,
- "spread_check": False}
+ "spread_check": False, "trim_hold": False}
return cls(**{**legacy, **(data or {})})
@classmethod
def v040(cls) -> "EnvelopeConfig":
"""The 0.4.0 rule set: everything before the spread check. The rule-based reference
controller runs with it so its 0.4.0 result reproduces exactly."""
- return cls(spread_check=False)
+ return cls(spread_check=False, trim_hold=False)
+
+ @classmethod
+ def v070(cls) -> "EnvelopeConfig":
+ """The 0.7.0 rule set: everything before the trim hold. Runs on prompt versions before 6
+ use it, so they behave as they did in 0.7.0."""
+ return cls(trim_hold=False)
def rate(self, device: str) -> float:
return {"cooling": self.cooling_rate_per_step, "dhw": self.dhw_rate_per_step,
@@ -203,6 +238,11 @@ def _r(v: float | None) -> float | None:
return None if v is None else round(float(v), 4)
+def _hold_event(b: str, action: str, depth_before: float, depth_after: float, reason: str) -> dict:
+ return {"building": b, "action": action, "depth_before": _r(depth_before), "depth_after": _r(depth_after),
+ "reason": reason}
+
+
@dataclass
class Decision:
"""Actions per building and device; None means the baseline action is used."""
@@ -211,6 +251,7 @@ class Decision:
clamps: list[Clamp] = field(default_factory=list)
fallback: str | None = None
reason: str = ""
+ holds: list[dict] = field(default_factory=list) # trim hold actions this hour (0.8.0)
@property
def status(self) -> str:
@@ -227,13 +268,16 @@ def resolved(self, ctx: StepContext) -> dict[str, dict[str, float]]:
return out
def as_dict(self) -> dict:
- return {
+ out = {
"status": self.status,
"reason": self.reason,
"fallback": self.fallback,
"actions": {b: {d: _r(v) for d, v in acts.items()} for b, acts in self.actions.items()},
"clamps": [c.as_dict() for c in self.clamps],
}
+ if self.holds: # absent when the hold did nothing, so older runs replay byte for byte
+ out["holds"] = [dict(h) for h in self.holds]
+ return out
def baseline_decision(buildings, fallback: str | None = None) -> Decision:
@@ -245,11 +289,13 @@ def __init__(self, config: EnvelopeConfig | None = None) -> None:
self.config = config or EnvelopeConfig()
self.memory: dict[str, float] = {} # "Building_1.battery" -> last applied action
self.trims: dict[str, float] = {} # building -> last hour's cooling minus its baseline (0.7.0, read only)
+ self.holds: dict[str, float] = {} # building -> depth of the cooling trim being held (0.8.0)
def fallback(self, ctx: StepContext, why: str) -> Decision:
"""Use the baseline action everywhere and remember it as the last action."""
self.memory = {f"{b}.{d}": st.baseline[d] for b, st in ctx.buildings.items() for d in DEVICE_KEYS}
self.trims = {}
+ self.holds = {}
return baseline_decision(ctx.buildings, fallback=why)
def apply(self, proposal: dict | None, ctx: StepContext, invalid_reason: str | None = None) -> Decision:
@@ -258,11 +304,16 @@ def apply(self, proposal: dict | None, ctx: StepContext, invalid_reason: str | N
clamps: list[Clamp] = []
actions: dict[str, dict[str, float | None]] = {}
memory: dict[str, float] = {}
+ events: list[dict] = []
+ holds: dict[str, float] = {}
+ early = proposal.get("end_trim") or {}
for b, st in ctx.buildings.items():
req = proposal["buildings"].get(b, {})
actions[b] = {}
for d in DEVICE_KEYS:
v = self._limit(b, d, req.get(d), st, clamps, ctx.prices)
+ if d == "cooling" and self.config.trim_hold:
+ v = self._hold_trim(b, v, st, early.get(b), clamps, events, holds)
actions[b][d] = v
memory[f"{b}.{d}"] = st.baseline[d] if v is None else v
if self.config.peak_limit:
@@ -270,7 +321,78 @@ def apply(self, proposal: dict | None, ctx: StepContext, invalid_reason: str | N
self.memory = memory
self.trims = {b: 0.0 if actions[b]["cooling"] is None else actions[b]["cooling"] - st.baseline["cooling"]
for b, st in ctx.buildings.items()}
- return Decision(actions=actions, clamps=clamps, reason=proposal["reason"])
+ if self.config.trim_hold:
+ self.holds = holds
+ return Decision(actions=actions, clamps=clamps, reason=proposal["reason"], holds=events)
+
+ def held_trim(self, b: str, st: BuildingState) -> tuple[float, str, str]:
+ """What the trim hold requires of building b this hour, before any early end:
+ (depth cooling must stay at least this far below the baseline, kind, why). kind is
+ "none" (no trim on), "end" (no trim may run now), "hold" or "ease"."""
+ cfg = self.config
+ prev = self.holds.get(b, 0.0)
+ gap = st.indoor_c - st.setpoint_c
+ if st.baseline["cooling"] <= cfg.trim_min_depth:
+ return 0.0, "end", "the baseline has no cooling to trim"
+ if not st.occupied:
+ return 0.0, "end", "the room is unoccupied"
+ if gap >= 0:
+ return 0.0, "end", f"indoor {st.indoor_c:.1f} C is at or above the setpoint {st.setpoint_c:.1f} C"
+ if prev <= cfg.trim_min_depth:
+ return 0.0, "none", ""
+ if gap < -cfg.trim_hold_c:
+ return prev, "hold", (f"indoor {st.indoor_c:.1f} C is more than {cfg.trim_hold_c:g} C below the setpoint "
+ f"{st.setpoint_c:.1f} C, trim held at {prev:.3f}")
+ need = round(max(0.0, prev - cfg.trim_release_step), 4)
+ return need, "ease", (f"indoor {st.indoor_c:.1f} C is within {cfg.trim_hold_c:g} C of the setpoint "
+ f"{st.setpoint_c:.1f} C, trim eases off from {prev:.3f} by at most "
+ f"{cfg.trim_release_step:g} per hour")
+
+ def _hold_trim(self, b: str, v: float | None, st: BuildingState, early: str | None, clamps: list[Clamp],
+ events: list[dict], holds: dict[str, float]) -> float | None:
+ """Carry building b's cooling trim (0.8.0). A trim is cooling more than trim_min_depth
+ below the baseline action, and its depth is how far below. This runs after the comfort
+ rule and only keeps cooling below the baseline, by no more than the trim already passed
+ the deviation limit with, so it never undoes another rule. Every action is recorded in
+ events; a changed value is also a clamp with rule trim_hold."""
+ cfg = self.config
+ point = f"{b}.cooling"
+ base = st.baseline["cooling"]
+ prev = self.holds.get(b, 0.0)
+ on = prev > cfg.trim_min_depth
+ applied = base if v is None else v
+ need, kind, why = self.held_trim(b, st)
+ if kind != "end" and on and early:
+ kind, why = "end_early", f"the model ended it early: {early}"
+ if kind in ("end", "end_early"):
+ trimmed = base - applied > cfg.trim_min_depth
+ if trimmed:
+ clamps.append(Clamp(point, "trim_hold", applied, round(base, 4),
+ f"trim ended, {why}, cooling back to the baseline"))
+ v = round(base, 4)
+ if on or trimmed:
+ events.append(_hold_event(b, kind, prev, 0.0, early if kind == "end_early" else why))
+ return v
+ acted = False
+ # A miss smaller than trim_min_depth is rounding (the observation shows the baseline and
+ # trims to 3 decimals) and is let through; the held depth below still counts in full.
+ if on and base - applied < need - cfg.trim_min_depth:
+ lo = max(cfg.cooling_min, st.bounds["cooling"][0])
+ nv = round(max(lo, base - need), 4)
+ clamps.append(Clamp(point, "trim_hold", applied, nv, why))
+ v = applied = nv
+ acted = True
+ depth = round(max(base - applied, need), 4)
+ if depth <= cfg.trim_min_depth:
+ if on:
+ events.append(_hold_event(b, "release", prev, 0.0, "trim eased off fully"))
+ return v
+ holds[b] = depth
+ if not on:
+ events.append(_hold_event(b, "start", 0.0, depth, "trim started"))
+ elif acted:
+ events.append(_hold_event(b, kind, prev, depth, why))
+ return v
def peak_target(self, ctx: StepContext) -> float | None:
if ctx.peak is None:
diff --git a/hyperbolic/live.py b/hyperbolic/live.py
index fc770f7..591e415 100644
--- a/hyperbolic/live.py
+++ b/hyperbolic/live.py
@@ -75,8 +75,10 @@
{"id": "week", "label": "1 week (167 hours)", "hours": 167},
{"id": "month", "label": "Full month (719 hours)", "hours": dc.EPISODE_HOURS},
]
-# Typical prompt version 5 request and reply, for a cost estimate before any run is recorded.
-TYPICAL_TOKENS = (3000, 250)
+# Typical prompt version 6 request and reply, for a cost estimate before any run is recorded: the
+# 0.7.0 live month averaged 4,946 input and 771 output tokens on version 5, and version 6 is about
+# 5 percent larger.
+TYPICAL_TOKENS = (5200, 790)
def key_status(env: dict[str, str] | None = None) -> dict:
diff --git a/hyperbolic/live_assets/live.js b/hyperbolic/live_assets/live.js
index 843a9eb..6c8f657 100644
--- a/hyperbolic/live_assets/live.js
+++ b/hyperbolic/live_assets/live.js
@@ -274,6 +274,13 @@
if (d.clamps.length > 3) ul.appendChild(el("li", "", "+" + (d.clamps.length - 3) + " more"));
li.appendChild(ul);
}
+ if (d.trim_ends && d.trim_ends.length) {
+ var te = el("ul", "trim-ends");
+ d.trim_ends.forEach(function (t) {
+ te.appendChild(el("li", "", t.building.replace("_", " ") + ": trim ended early: " + t.reason));
+ });
+ li.appendChild(te);
+ }
ui.log.insertBefore(li, ui.log.firstChild);
while (ui.log.childNodes.length > LOG_MAX) ui.log.removeChild(ui.log.lastChild);
}
diff --git a/hyperbolic/model.py b/hyperbolic/model.py
index f4c29e9..64d86ec 100644
--- a/hyperbolic/model.py
+++ b/hyperbolic/model.py
@@ -4,6 +4,10 @@
ScriptedSupervisor returns deterministic rule-based replies for offline runs and tests.
HoldSupervisor always keeps the baseline action (the live page's baseline option).
ReplaySupervisor serves recorded replies from a run directory, with zero API calls.
+RecordedReplySupervisor
+ serves a recorded run's replies hour by hour whatever the request, so the
+ same answers can be sent through other envelope rules (an evaluation, never
+ a recorded run).
"""
from __future__ import annotations
@@ -184,6 +188,32 @@ def call(self, step: int, request: dict) -> CallResult:
)
+class RecordedReplySupervisor:
+ """Answers each hour with the reply a recorded run got for that hour, without checking the
+ request. Used to send a live run's own actions through different envelope rules
+ (scripts/evaluate_hold.py); with the run's own rules it reproduces the run exactly."""
+
+ kind = "recorded"
+
+ def __init__(self, calls_path: Path, model_key: str) -> None:
+ self.model_key = model_key
+ self._inner = ReplaySupervisor(calls_path, model_key)
+
+ def call(self, step: int, request: dict) -> CallResult:
+ rec = self._inner._by_step.get(step)
+ if rec is None:
+ return CallResult(None, error="no recorded reply for this hour")
+ reply = rec["reply"]
+ return CallResult(
+ text=reply.get("text"),
+ input_tokens=reply.get("input_tokens", 0),
+ output_tokens=reply.get("output_tokens", 0),
+ stop_reason=reply.get("stop_reason"),
+ error=reply.get("error"),
+ model=reply.get("model"),
+ )
+
+
@dataclass(frozen=True)
class RuleParams:
"""Knobs of the rule-based policy for prompt version 3 observations. Defaults are the
@@ -279,7 +309,8 @@ class ScriptedSupervisor:
building's own load, and can shave the district peak: the 0.4.0 rule-based controller.
For version 4 it follows that prompt's strategy literally (strategy_plan_v4), which is
how the strategy's numbers are chosen offline, and likewise for version 5
- (strategy_plan_v5). Version 2 observations get the 0.3.0
+ (strategy_plan_v5). Version 6 runs the same version 5 strategy, never ends a trim early,
+ and leaves holding trims to the envelope. Version 2 observations get the 0.3.0
rule and version 1 the simpler 0.2.0 rule.
Faults are opt-in: on listed steps it injects a bad reply so the envelope and fallback
@@ -321,6 +352,9 @@ def call(self, step: int, request: dict) -> CallResult:
stop_reason="end_turn", model="scripted")
if "smoothing" in obs:
reply = strategy_plan_v5(obs, self.strategy_v5)
+ if "strategy_battery_reason" in obs: # version 6: the reply carries end_trim
+ for b in obs["buildings"]:
+ reply[b]["end_trim"] = None
elif "price_plan" in obs:
reply = strategy_plan_v4(obs, self.strategy)
elif "signal_outlook" in obs:
@@ -567,6 +601,22 @@ def strategy_plan_v5(obs: dict, p: StrategyParamsV5 | None = None) -> dict:
started, is held until the room is nearly back at its setpoint and then released a little each
hour; a grid charge starts small and rises a little each hour; the top-price discharge can keep
some charge back for the evening district peak."""
+ reply, battery_notes, cooling_notes = _strategy_v5(obs, p)
+ notes = battery_notes | cooling_notes
+ reason = "; ".join(sorted(notes)) if notes else "Nothing stands out, hold storage and meet the cooling load"
+ return {"reason": (reason[0].upper() + reason[1:])[:dc.REASON_TARGET_CHARS], **reply}
+
+
+def battery_plan_v5(obs: dict, p: StrategyParamsV5 | None = None) -> tuple[dict[str, float], str]:
+ """The version 5 strategy's battery action for each building and why: what prompt version 6
+ shows the model each hour as strategy_battery, in place of the 0.4.0 rule-based plan."""
+ reply, battery_notes, _ = _strategy_v5(obs, p)
+ reason = "; ".join(sorted(battery_notes)) if battery_notes else "nothing stands out, hold the batteries"
+ return {b: r["battery"] for b, r in reply.items()}, (reason[0].upper() + reason[1:])[:dc.REASON_TARGET_CHARS]
+
+
+def _strategy_v5(obs: dict, p: StrategyParamsV5 | None = None) -> tuple[dict, set[str], set[str]]:
+ """strategy_plan_v5's actions per building, with its battery and cooling notes kept apart."""
p = p or StrategyParamsV5()
pp, peak, nxt = obs["price_plan"], obs["peak"], obs["next_24h"]
top_now = bool(pp["top_price_now"])
@@ -586,7 +636,8 @@ def strategy_plan_v5(obs: dict, p: StrategyParamsV5 | None = None) -> dict:
caps = {b: s["battery"]["capacity_kwh"] or 0.0 for b, s in obs["buildings"].items()}
total_cap = sum(caps.values()) or 1.0
reply: dict = {}
- notes = set()
+ notes: set[str] = set() # battery notes
+ cooling_notes: set[str] = set()
for b, s in sorted(obs["buildings"].items()):
bat = s["battery"]
cap, fmax, soc = caps[b], bat["max_fraction_per_hour"] or 0.0, bat["soc"]
@@ -636,12 +687,11 @@ def strategy_plan_v5(obs: dict, p: StrategyParamsV5 | None = None) -> dict:
# The baseline arrives rounded to 3 decimals: keep 0.001 clear of the deviation limit.
cooling = math.ceil(max(0.0, base + max(trim, -p.cold_trim) + 0.001) * 1000) / 1000
if cooling < base:
- notes.add("hold a cooling trim in a cold room, easing it off slowly")
+ cooling_notes.add("hold a cooling trim in a cold room, easing it off slowly")
else:
cooling = None
reply[b] = {"cooling": cooling, "dhw": 0.0, "battery": math.trunc(battery * 1000) / 1000}
- reason = "; ".join(sorted(notes)) if notes else "Nothing stands out, hold storage and meet the cooling load"
- return {"reason": (reason[0].upper() + reason[1:])[:dc.REASON_TARGET_CHARS], **reply}
+ return reply, notes, cooling_notes
class HoldSupervisor:
diff --git a/hyperbolic/prompt.py b/hyperbolic/prompt.py
index f8f78a8..06eb253 100644
--- a/hyperbolic/prompt.py
+++ b/hyperbolic/prompt.py
@@ -21,6 +21,12 @@
cooling trim last hour. The strategy holds a trim until the
room is nearly back at its setpoint and eases it off slowly, starts grid charging
small and raises it a little each hour, and spreads discharge over the top-price hours.
+ 6 v0.8.0. The envelope holds a cooling trim in code (EnvelopeConfig.trim_hold), and the
+ prompt says so: null keeps a held trim, and the reply's new end_trim field, with a
+ reason, is the only way to end one early. The 0.4.0 rule-based reference_action is
+ replaced by the version 5 strategy's own battery action per building
+ (strategy_battery, with strategy_battery_reason), and each building shows the trim the
+ envelope will keep this hour (cooling_trim_held).
"""
from __future__ import annotations
@@ -29,12 +35,27 @@
from . import district as dc
from .envelope import Decision, Envelope, EnvelopeConfig, StepContext, proposal_json_schema
-from .model import DEFAULT_EFFORT, DEFAULT_MAX_TOKENS, MODELS, StrategyParams, StrategyParamsV5, rule_plan_v3
+from .model import (
+ DEFAULT_EFFORT,
+ DEFAULT_MAX_TOKENS,
+ MODELS,
+ StrategyParams,
+ StrategyParamsV5,
+ battery_plan_v5,
+ rule_plan_v3,
+)
from .signals import charge_spread, combined_signal
-PROMPT_VERSION = 5
-PROMPT_VERSIONS = (1, 2, 3, 4, 5)
+PROMPT_VERSION = 6
+PROMPT_VERSIONS = (1, 2, 3, 4, 5, 6)
+# From this version the reply may end a held cooling trim early (envelope trim hold, 0.8.0).
+END_TRIM_FROM = 6
+
+
+def end_trim_field(version: int) -> bool:
+ """Whether replies to this prompt version carry end_trim per building."""
+ return version >= END_TRIM_FROM
def _r(v, nd=3):
@@ -599,6 +620,108 @@ def system_prompt_v5(cfg: EnvelopeConfig, buildings: tuple[str, ...]) -> str:
reply invalid."""
+def build_observation_v6(state: dict, ctx: StepContext, last: Decision | None, look: Lookahead,
+ envelope: Envelope | None) -> dict:
+ """Version 6: version 5 with the version 5 strategy's battery plan in place of the 0.4.0
+ rule-based reference action, and the trim the envelope will hold this hour."""
+ obs = build_observation_v5(state, ctx, last, look, envelope)
+ del obs["reference_reason"]
+ for b in obs["buildings"]:
+ del obs["buildings"][b]["reference_action"]
+ battery, why = battery_plan_v5(obs)
+ obs["strategy_battery_reason"] = why
+ for b, st in ctx.buildings.items():
+ held = 0.0 if envelope is None or not envelope.config.trim_hold else envelope.held_trim(b, st)[0]
+ obs["buildings"][b]["strategy_battery"] = battery[b]
+ obs["buildings"][b]["cooling_trim_held"] = _r(-held, 3)
+ return obs
+
+
+def system_prompt_v6(cfg: EnvelopeConfig, buildings: tuple[str, ...]) -> str:
+ dev = ("" if cfg.cooling_max_deviation is None else
+ f"\n- cooling more than {cfg.cooling_max_deviation:g} away from the baseline_action")
+ guard = ("\n- extra cooling in an unoccupied building already near the bottom of its band"
+ if cfg.overcool_guard_unoccupied else "")
+ peak = ("\n- battery then tank charging, so district load stays at or below peak.target_kwh"
+ if cfg.peak_limit else "")
+ spread = ("\n- grid charging of a battery when grid_charging_ok is false (spare solar may still be stored)"
+ if cfg.spread_check else "")
+ hold = "\n- cooling above a held trim, and a trim where the hold has ended" if cfg.trim_hold else ""
+ sp = StrategyParamsV5() # the strategy's numbers, chosen on the tuning dataset
+ full = sp.cold_margin_c + sp.cold_ramp_c
+ reserve = ("" if sp.evening_reserve <= 0 else
+ f" Keep {sp.evening_reserve:g} of state of charge back for the evening district peak.")
+ return f"""You are the hourly supervisory controller for a simulated district of {len(buildings)} \
+buildings ({dc.DATASET_NAME}, CityLearn). Each building has a heat pump for space cooling, a hot water \
+tank, a battery and rooftop solar. Every hour you choose one action per device for every building.
+
+Goals, in this order, graded by CityLearn's own KPIs over the month against the no-control reference \
+(storage idle, heat pump meeting the ideal load):
+1. Comfort: no more discomfort than the reference. Discomfort is any occupied hour with indoor \
+temperature outside setpoint +/- {cfg.comfort_band_k:g} C.
+2. Peaks and smoothness: keep the district's daily peak, its daily load factor and its ramping at or \
+better than the reference. Ramping counts every hour the district draws more than the hour before, \
+so a step up costs and a step down does not.
+3. Cost first, then carbon: lower electricity cost and lower emissions than the reference. Each \
+building's cost counts equally, whatever its size.
+
+What you see:
+- strategy_battery in each building: the battery action points 2 to 4 below give this hour, worked \
+out in code, with strategy_battery_reason. Use it unless you have a clear reason; say why you differ.
+- price_plan: the price now, the 24 hour top and low, top_price_now (this is one of the most \
+expensive hours), top_price_hours_ahead (hours ahead at the top price) and flat_24h (no price \
+difference in view, as on weekends).
+- battery: soc, soc_floor (the battery cannot go below it), usable_kwh above the floor, capacity, \
+max_fraction_per_hour, round_trip_min (the share of a charge that comes back at worst) and \
+grid_charging_ok (false when no price in the next 24 hours pays back the losses and a cheaper \
+hour is still coming).
+- indoor_c is the measured room temperature; indoor_vs_setpoint_c is how far it is from the \
+setpoint; cooling_trim_last_hour is how far below baseline_action cooling ran last hour; \
+cooling_trim_held is the trim the envelope keeps this hour.
+- smoothing: district_draw_last_hour_kwh (measured), district_load_now_kwh (before storage), \
+district_load_next_hour_kwh (the next hour's load before storage, as yesterday).
+- peak: target_kwh, headroom_kwh, forecast_today_kwh.
+- signal_outlook and next_24h: price and carbon by hour. Carbon is the same hour yesterday.
+
+Strategy:
+1. Comfort and cooling. An occupied room more than {sp.cold_margin_c:g} C below its setpoint is \
+overcooled: cut its cooling below baseline_action, from 0 at {sp.cold_margin_c:g} C below the \
+setpoint to {sp.cold_trim:g} at {full:g} C or more below. The envelope holds a trim in code: while \
+the room is occupied and more than {cfg.trim_hold_c:g} C below its setpoint it stays at least as deep \
+as last hour, within {cfg.trim_hold_c:g} C it eases off by at most {cfg.trim_release_step:g} per hour, \
+and it ends at the setpoint, when the room empties or when baseline cooling is 0. null keeps a held \
+trim; you may trim deeper. End one early only through end_trim; early ends are logged.
+2. Hold the battery for the top-price hours. When top_price_now is true, discharge the usable charge \
+evenly over the top-price hours left, each hour no more than the building's load_now_kwh (power \
+sent back to the grid earns nothing).{reserve} Do not discharge in cheaper hours before them, except \
+to keep district load down: once district_load_now_kwh reaches {sp.shave:g} x forecast_today_kwh, \
+discharge to cover the building's own load. On flat_24h days leave the battery idle.
+3. Charge so each battery is near {sp.soc_target:g} when the top-price hours start: only when \
+grid_charging_ok is true, in the cheapest hours with the lowest carbon (usually the early morning), \
+up to about {sp.charge_rate:g} of capacity per hour, starting small and rising by at most \
+{sp.charge_step:g} per hour, within headroom_kwh. Store spare solar any time.
+4. Never charge at the top price. Leave the hot water tank at 0.
+
+Actions per building, use null to keep the baseline action:
+- cooling: heat pump power as a fraction of nominal, {cfg.cooling_min:g} to {cfg.cooling_max:g}.
+- dhw: hot water tank, fraction of capacity to charge (+) or discharge (-) this hour. Baseline 0.
+- battery: fraction of battery capacity to charge (+) or discharge (-) this hour. Baseline 0.
+- end_trim: null, or a short reason to end that building's held cooling trim now.
+
+A safety envelope checks every value before it is applied. It will clamp:
+- any value outside its range or CityLearn's action space
+- any move larger than {cfg.cooling_rate_per_step:g} (cooling), {cfg.dhw_rate_per_step:g} (dhw) or \
+{cfg.battery_rate_per_step:g} (battery) per hour, measured from last hour's action or this hour's baseline
+- charge that would leave a battery or tank outside {cfg.soc_min:g} to {cfg.soc_max:g} state of charge{dev}
+- in occupied hours within {cfg.comfort_margin_k:g} C of a comfort band edge, cooling below the baseline \
+when warm or above it when cool{guard}{hold}{spread}{peak}
+An invalid reply uses the baseline action everywhere for that hour. Clamps are logged against you.
+
+Reply with one JSON object only, matching the schema. "reason" is one plain sentence of at most \
+{dc.REASON_TARGET_CHARS} characters. A reason over {dc.REASON_MAX_CHARS} characters makes the whole \
+reply invalid."""
+
+
def observe(version: int, state: dict, ctx: StepContext, last: Decision | None, look: Lookahead,
envelope: Envelope | None = None) -> dict:
if version == 1:
@@ -611,6 +734,8 @@ def observe(version: int, state: dict, ctx: StepContext, last: Decision | None,
return build_observation_v4(state, ctx, last, look, envelope)
if version == 5:
return build_observation_v5(state, ctx, last, look, envelope)
+ if version == 6:
+ return build_observation_v6(state, ctx, last, look, envelope)
raise ValueError(f"unknown prompt version {version}")
@@ -625,7 +750,7 @@ def build_request(obs: dict, cfg: EnvelopeConfig, buildings: tuple[str, ...], mo
max_tokens: int = DEFAULT_MAX_TOKENS, effort: str = DEFAULT_EFFORT,
prompt_version: int = 1) -> dict:
system = {1: system_prompt, 2: system_prompt_v2, 3: system_prompt_v3,
- 4: system_prompt_v4, 5: system_prompt_v5}[prompt_version](cfg, buildings)
+ 4: system_prompt_v4, 5: system_prompt_v5, 6: system_prompt_v6}[prompt_version](cfg, buildings)
return {
"model": MODELS[model_key]["id"],
"max_tokens": max_tokens,
@@ -633,6 +758,7 @@ def build_request(obs: dict, cfg: EnvelopeConfig, buildings: tuple[str, ...], mo
"messages": [{"role": "user", "content": user_message(obs)}],
"output_config": {
"effort": effort,
- "format": {"type": "json_schema", "schema": proposal_json_schema(buildings)},
+ "format": {"type": "json_schema",
+ "schema": proposal_json_schema(buildings, end_trim_field(prompt_version))},
},
}
diff --git a/hyperbolic/runner.py b/hyperbolic/runner.py
index 3596521..41210c2 100644
--- a/hyperbolic/runner.py
+++ b/hyperbolic/runner.py
@@ -29,22 +29,27 @@
request_sha256,
worst_case_cost_usd,
)
-from .prompt import PROMPT_VERSION, Lookahead, build_request, observe
+from .prompt import PROMPT_VERSION, Lookahead, build_request, end_trim_field, observe
from .signals import PeakInfo, PeakTracker, PriceOutlook
# The rule-based controller is the 0.4.0 reference, kept frozen so its result reproduces: prompt
-# version 3 and the 0.4.0 envelope. Prompt version 4 shows its plan to the model each hour.
+# version 3 and the 0.4.0 envelope. Prompt versions 4 and 5 show its plan to the model each hour.
RULE_PROMPT_VERSION = 3
+# Prompt version 6 is the first that knows the envelope holds cooling trims (0.8.0).
+HOLD_PROMPT_VERSION = 6
def controller_settings(rule_based: bool, prompt_version: int | None = None) -> tuple[int, EnvelopeConfig]:
"""Prompt version and envelope rules for a new run. Without a prompt version, models get the
- current one and the rule-based controller the 0.4.0 one, each with its own envelope."""
+ current one and the rule-based controller the 0.4.0 one, each with its own envelope. Prompt
+ versions before 6 run with the 0.7.0 rules, as they did then; version 6 adds the trim hold."""
if prompt_version is None:
prompt_version = RULE_PROMPT_VERSION if rule_based else PROMPT_VERSION
if rule_based and prompt_version <= RULE_PROMPT_VERSION:
return prompt_version, EnvelopeConfig.v040()
+ if prompt_version < HOLD_PROMPT_VERSION:
+ return prompt_version, EnvelopeConfig.v070()
return prompt_version, EnvelopeConfig()
@@ -228,7 +233,7 @@ def decide(self, step: int, state: dict, writer: RunWriter) -> tuple[Decision, d
if res.error:
decision = self.envelope.fallback(ctx, f"no usable reply: {res.error}")
else:
- proposal, err = parse_reply(res.text, buildings)
+ proposal, err = parse_reply(res.text, buildings, end_trim_field(self.prompt_version))
decision = self.envelope.apply(proposal, ctx, None if err is None else f"invalid reply: {err}")
if decision.status == "clamped":
self.counts["clamped"] += 1
diff --git a/hyperbolic/viewer.py b/hyperbolic/viewer.py
index 188ab11..5a3abe2 100644
--- a/hyperbolic/viewer.py
+++ b/hyperbolic/viewer.py
@@ -88,19 +88,33 @@ def _actions(run: dict, n: int, buildings: list[str]) -> dict:
def _decision(s: dict) -> dict:
d = s.get("decision") or {}
- return {
+ out = {
"status": d.get("status", "ok"),
"reason": d.get("reason") or "",
"fallback": d.get("fallback"),
"clamps": [{"point": c["point"], "rule": c["rule"], "requested": c["requested"],
"applied": c["applied"], "reason": c["reason"]} for c in d.get("clamps") or []],
}
+ # Cooling trims the model ended early, with its reason (0.8.0). Present only when there are
+ # some, so the viewer data of older runs is unchanged.
+ ends = [{"building": h["building"], "reason": h["reason"]} for h in d.get("holds") or []
+ if h.get("action") == "end_early"]
+ if ends:
+ out["trim_ends"] = ends
+ return out
def _decisions(run: dict, n: int) -> list[dict]:
return [_decision(s) for s in run["steps"][:n]]
+def _trim_end_count(manifest: dict, decisions: list[dict]) -> dict:
+ """How many cooling trims the model ended early, for runs recorded with the trim hold."""
+ if not (manifest.get("envelope") or {}).get("trim_hold"):
+ return {}
+ return {"trim_ends_early": sum(len(d.get("trim_ends", [])) for d in decisions)}
+
+
def _pct(agent: float | None, base: float | None) -> float | None:
if agent is None or base is None or abs(base) < 1e-12:
return None
@@ -184,6 +198,7 @@ def build_scenario(baseline_dir: Path, agent_dir: Path) -> dict:
"fallback_hours": sum(1 for d in decisions if d["status"] == "fallback"),
"budget_stopped": bool(ar.get("budget_stopped")),
"envelope": am.get("envelope"),
+ **_trim_end_count(am, decisions),
},
"headline": headline(base["result"].get("district_kpis") or {}, ar.get("district_kpis") or {}),
}
diff --git a/pyproject.toml b/pyproject.toml
index 5de6b05..3e31ec1 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
[project]
name = "hyperbolic"
-version = "0.7.0"
+version = "0.8.0"
description = "Supervisory control for CityLearn district simulations, behind a deterministic safety envelope."
readme = "README.md"
license = { text = "MIT" }
diff --git a/scripts/evaluate_hold.py b/scripts/evaluate_hold.py
new file mode 100644
index 0000000..dc956ea
--- /dev/null
+++ b/scripts/evaluate_hold.py
@@ -0,0 +1,159 @@
+"""Estimate the 0.8.0 trim hold on a recorded live run, as an evaluation, never a recorded run.
+
+ python scripts/evaluate_hold.py RUN_DIR
+
+The run's own replies are sent hour by hour, unchanged, through two envelopes on a fresh
+CityLearn month with zero model calls:
+
+ control the rules the run was recorded with. It must reproduce the run's published KPIs
+ exactly, or the script stops: that proves the replay path itself changes nothing.
+ hold the same rules with the trim hold on (EnvelopeConfig.trim_hold).
+
+The replies are fixed, so later hours do not react to what the hold changed (a warmer or cooler
+room), and nothing here measures what a version 6 prompt would change. It prints both sets of
+KPIs, the clamps by rule, the hold's actions, trim episodes counted as in the 0.7.0 diagnosis,
+and the exact ramping split (scripts/split_ramping.py) for both.
+"""
+
+from __future__ import annotations
+
+import collections
+import dataclasses
+import json
+import statistics
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+sys.path.insert(0, str(ROOT / "scripts"))
+
+from split_ramping import split, table # noqa: E402
+
+from hyperbolic.envelope import Envelope, EnvelopeConfig # noqa: E402
+from hyperbolic.model import Budget, RecordedReplySupervisor # noqa: E402
+from hyperbolic.runner import AgentController, NullWriter, load_jsonl, run_session # noqa: E402
+
+KPI_KEYS = ("cost_total", "carbon_emissions_total", "electricity_consumption_total", "discomfort_proportion",
+ "discomfort_cold_proportion", "discomfort_hot_proportion", "daily_peak_average", "all_time_peak_average",
+ "ramping_average", "daily_one_minus_load_factor_average", "power_outage_normalized_unserved_energy_total")
+TRIM_MIN = 0.002
+
+
+def evaluate(run_dir: Path, cfg: EnvelopeConfig) -> dict:
+ from hyperbolic.sim import CityLearnSimulator
+
+ class KeepEnv(CityLearnSimulator):
+ def close(self) -> None: # keep the finished env for the ramping split
+ pass
+
+ m = json.loads((run_dir / "manifest.json").read_text(encoding="utf-8"))
+ sup = m["supervisor"]
+ ctl = AgentController(
+ supervisor=RecordedReplySupervisor(run_dir / "calls.jsonl", sup["model_key"]),
+ envelope=Envelope(cfg),
+ budget=Budget(limit_usd=float(sup["budget_usd"])),
+ max_tokens=int(sup["max_tokens"]),
+ effort=sup["effort"],
+ prompt_version=int(sup.get("prompt_version", 1)),
+ )
+ sim, w = KeepEnv(control=True), NullWriter()
+ res = run_session(sim, ctl, w, hours=m.get("hours_limit"), sim_info={"kind": "citylearn", "control": True},
+ log=lambda *_: None)
+ return {"result": res, "steps": w.steps, "split": split(sim.env, w.steps)}
+
+
+def trim_episodes(steps: list[dict]) -> dict:
+ """Trim episodes as the 0.7.0 diagnosis counted them: cooling more than 0.002 below the
+ baseline action, and episodes that ended within the month. An end broke the version 5 rule
+ when the room was occupied, still below its setpoint and the baseline still had cooling to
+ trim, and either the room was more than 0.5 C too cold or the trim dropped more than 0.02."""
+ lengths, occupied_ends, cold_ends, fast_ends = [], 0, 0, 0
+ for b in steps[0]["actions"]:
+ run, depth = 0, 0.0
+ for s in steps:
+ d = s["state"]["baseline"][b]["cooling"] - s["actions"][b]["cooling"]
+ if d > TRIM_MIN:
+ run, depth = run + 1, d
+ continue
+ if run:
+ lengths.append(run)
+ st = s["state"]["buildings"][b]
+ gap = st["indoor_c"] - st["setpoint_c"]
+ if st["occupants"] > 0:
+ occupied_ends += 1
+ if gap < 0 and s["state"]["baseline"][b]["cooling"] > TRIM_MIN:
+ if gap < -0.5:
+ cold_ends += 1
+ elif depth > 0.02 + 1e-9:
+ fast_ends += 1
+ run = 0
+ return {"episodes": len(lengths), "median_hours": statistics.median(lengths) if lengths else 0,
+ "one_hour": sum(1 for x in lengths if x == 1), "ended_occupied": occupied_ends,
+ "broke_rule": cold_ends + fast_ends, "ended_too_cold": cold_ends, "dropped_fast": fast_ends}
+
+
+def clamp_rules(steps: list[dict]) -> dict:
+ hours, clamps = collections.Counter(), collections.Counter()
+ for s in steps:
+ rules = [c["rule"] for c in (s["decision"] or {}).get("clamps", [])]
+ clamps.update(rules)
+ hours.update(set(rules))
+ return {r: (hours[r], clamps[r]) for r in sorted(clamps)}
+
+
+def hold_actions(steps: list[dict]) -> dict:
+ out = collections.Counter()
+ for s in steps:
+ for h in (s["decision"] or {}).get("holds", []):
+ kind = h["action"]
+ if kind == "end":
+ kind = ("end: no baseline cooling" if "baseline" in h["reason"] else
+ "end: unoccupied" if "unoccupied" in h["reason"] else "end: at or above setpoint")
+ out[kind] += 1
+ return dict(sorted(out.items()))
+
+
+def main(argv: list[str] | None = None) -> int:
+ args = argv if argv is not None else sys.argv[1:]
+ if len(args) != 1:
+ print(__doc__.splitlines()[2].strip(), file=sys.stderr)
+ return 2
+ run_dir = Path(args[0])
+ m = json.loads((run_dir / "manifest.json").read_text(encoding="utf-8"))
+ published = json.loads((run_dir / "result.json").read_text(encoding="utf-8"))["district_kpis"]
+ control_cfg = EnvelopeConfig.from_manifest(m["envelope"])
+ if control_cfg.trim_hold:
+ print("this run was recorded with the trim hold already on", file=sys.stderr)
+ return 2
+ control = evaluate(run_dir, control_cfg)
+ got = control["result"]["district_kpis"]
+ if got != published:
+ diff = {k: (published.get(k), got.get(k)) for k in set(published) | set(got) if published.get(k) != got.get(k)}
+ print(f"control does not reproduce the published KPIs: {diff}", file=sys.stderr)
+ return 1
+ hold = evaluate(run_dir, dataclasses.replace(control_cfg, trim_hold=True))
+ recorded = load_jsonl(run_dir / "steps.jsonl")
+ if [s["actions"] for s in control["steps"]] != [s["actions"] for s in recorded]:
+ print("control actions differ from the recording", file=sys.stderr)
+ return 1
+ hk = hold["result"]["district_kpis"]
+ print(f"{run_dir.name}: its own replies through the {m.get('tool_version')} rules (control) and with the "
+ "trim hold (evaluation, not a recorded run)")
+ print("control reproduces the published KPIs exactly: yes")
+ print(f" {'CityLearn KPI (district)':<46}{'control':>10}{'hold':>10}{'change':>10}")
+ for k in KPI_KEYS:
+ print(f" {k:<46}{got[k]:>10.4f}{hk[k]:>10.4f}{hk[k] - got[k]:>+10.4f}")
+ for name, ev in (("control", control), ("hold", hold)):
+ r = ev["result"]
+ print(f" {name}: clamped hours {r['clamped_steps']}, fallback hours {r['fallback_steps']}; "
+ f"clamps by rule (hours, clamps) {clamp_rules(ev['steps'])}")
+ print(f" {name}: trims {trim_episodes(ev['steps'])}")
+ print(f" hold actions: {hold_actions(hold['steps'])}")
+ print()
+ print(table({"control": control["split"], "hold": hold["split"]}))
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/scripts/split_ramping.py b/scripts/split_ramping.py
new file mode 100644
index 0000000..6c776b8
--- /dev/null
+++ b/scripts/split_ramping.py
@@ -0,0 +1,159 @@
+"""Split CityLearn's district ramping by cause, exactly (the 0.8.0 diagnosis).
+
+ python scripts/split_ramping.py RUN_DIR [RUN_DIR ...]
+
+CityLearn's ramping_average is R(E) / R(E_ref). R adds up every hour-to-hour rise of a series
+(a fall counts for nothing), E is the district's net draw, and E_ref is CityLearn's own
+reference for the same run: its net draw with storage taken out and cooling put back to the
+ideal load (net_electricity_consumption_without_storage_and_partial_load). It is not the
+baseline run: in an outage hour E is 0 whatever the actions, but E_ref still has the cooling
+difference taken off, so extra cooling in an outage lowers E_ref there and adds a rise to it.
+
+Each run is simulated again from its recorded actions (the outcomes must match the recording).
+Per building, by construction, net = ref + battery + hot water storage + cooling gap, where the
+cooling gap is (net - storage) - ref. Each hour's rise max(0, dE) is split among the causes
+below with Shapley values, starting from the reference's own change that hour, so the shares
+add up exactly: R(E) = R(E_ref) + sum of the shares.
+
+ battery_charge, battery_discharge the positive and negative parts of the battery's draw
+ dhw_storage the hot water tank's draw
+ trim_start, trim_end, trim_depth the cooling gap while cooling is more than 0.002 below the
+ baseline action: the hour a trim starts, the hour it ends,
+ and changes in its depth in between
+ extra_cooling the cooling gap while cooling is more than 0.002 above it
+ cooling_rounding the remaining cooling gap (actions rounded to 4 decimals)
+ outage_hours everything in a building's outage hours, where the grid
+ draw is 0 whatever the actions
+"""
+
+from __future__ import annotations
+
+import itertools
+import json
+import math
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(ROOT))
+
+import numpy as np # noqa: E402
+
+from hyperbolic.runner import load_jsonl # noqa: E402
+
+CAUSES = ("battery_charge", "battery_discharge", "dhw_storage", "trim_start", "trim_end", "trim_depth",
+ "extra_cooling", "cooling_rounding", "outage_hours")
+TRIM_MIN = 0.002 # a trim is cooling more than this below the baseline action (EnvelopeConfig.trim_min_depth)
+
+
+def resimulate(run_dir: Path):
+ """Run a recorded run's actions through CityLearn again. Returns (CityLearnEnv, steps)."""
+ from hyperbolic.sim import CityLearnSimulator
+
+ steps = load_jsonl(Path(run_dir) / "steps.jsonl")
+ m = json.loads((Path(run_dir) / "manifest.json").read_text(encoding="utf-8"))
+ sim = CityLearnSimulator(control=steps[0]["actions"] is not None)
+ state = sim.start(m.get("hours_limit"))
+ i = 0
+ while state is not None and i < len(steps):
+ outcome, state = sim.step(steps[i]["actions"])
+ if outcome != steps[i]["outcome"]:
+ raise ValueError(f"{run_dir}: hour {i} does not reproduce the recorded outcome")
+ i += 1
+ return sim.env, steps
+
+
+def _rise(x: np.ndarray) -> float:
+ return float(np.clip(np.diff(x), 0.0, None).sum())
+
+
+def _shapley(start: float, deltas: list[float]) -> list[float]:
+ """Exact Shapley shares of max(0, start + sum(deltas)) - max(0, start)."""
+ phi = [0.0] * len(deltas)
+ live = [i for i, d in enumerate(deltas) if abs(d) > 1e-12]
+ m = len(live)
+ for i in live:
+ others = [j for j in live if j != i]
+ for r in range(m):
+ w = math.factorial(r) * math.factorial(m - r - 1) / math.factorial(m)
+ for sub in itertools.combinations(others, r):
+ s = start + sum(deltas[j] for j in sub)
+ phi[i] += w * (max(0.0, s + deltas[i]) - max(0.0, s))
+ return phi
+
+
+def split(env, steps: list[dict]) -> dict:
+ """The exact split for a finished CityLearn env and the run's step records."""
+ n = len(steps)
+ ref_total = np.zeros(n)
+ net_total = np.zeros(n)
+ inc = {c: np.zeros(n) for c in CAUSES}
+ for b in env.buildings:
+ name = b.name
+ net = np.asarray(b.net_electricity_consumption, dtype=float)[:n]
+ ws = np.asarray(b.net_electricity_consumption_without_storage, dtype=float)[:n]
+ ref = np.asarray(b.net_electricity_consumption_without_storage_and_partial_load, dtype=float)[:n]
+ bat = np.asarray(b.electrical_storage_electricity_consumption, dtype=float)[:n]
+ dhw = net - ws - bat # all other storage: the hot water tank here
+ gap = ws - ref
+ act = np.array([s["actions"][name]["cooling"] for s in steps])
+ base = np.array([s["state"]["baseline"][name]["cooling"] for s in steps])
+ out = np.array([bool(s["state"]["buildings"][name].get("power_outage")) for s in steps])
+ trimmed, extra = (act < base - TRIM_MIN) & ~out, (act > base + TRIM_MIN) & ~out
+ level = {
+ "battery_charge": np.where(out, 0.0, np.clip(bat, 0.0, None)),
+ "battery_discharge": np.where(out, 0.0, np.clip(bat, None, 0.0)),
+ "dhw_storage": np.where(out, 0.0, dhw),
+ "extra_cooling": np.where(extra, gap, 0.0),
+ "cooling_rounding": np.where(~trimmed & ~extra & ~out, gap, 0.0),
+ "outage_hours": np.where(out, bat + dhw + gap, 0.0),
+ }
+ for c, x in level.items():
+ inc[c][1:] += np.diff(x)
+ trim = np.where(trimmed, gap, 0.0)
+ d = np.diff(trim)
+ for t in range(1, n):
+ kind = ("trim_depth" if trimmed[t] and trimmed[t - 1] else "trim_start" if trimmed[t]
+ else "trim_end" if trimmed[t - 1] else None)
+ if kind:
+ inc[kind][t] += d[t - 1]
+ ref_total += ref
+ net_total += net
+ dref = np.zeros(n)
+ dref[1:] = np.diff(ref_total)
+ shares = dict.fromkeys(CAUSES, 0.0)
+ for t in range(1, n):
+ for c, v in zip(CAUSES, _shapley(dref[t], [inc[c][t] for c in CAUSES])):
+ shares[c] += v
+ r, r_ref = _rise(net_total), _rise(ref_total)
+ if abs(r_ref + sum(shares.values()) - r) > 1e-6:
+ raise AssertionError("the shares do not add up to the run's ramping")
+ return {"ramping_kwh": r, "reference_kwh": r_ref, "ramping_average": r / r_ref, "shares_kwh": shares}
+
+
+def table(results: dict[str, dict]) -> str:
+ names = list(results)
+ lines = [f"{'cause, points of ramping_average':<34}" + "".join(f"{n:>18}" for n in names)]
+ for c in CAUSES:
+ lines.append(f"{c:<34}" + "".join(f"{results[n]['shares_kwh'][c] / results[n]['reference_kwh'] * 100:>+18.2f}"
+ for n in names))
+ lines.append(f"{'reference rises (kWh)':<34}" + "".join(f"{results[n]['reference_kwh']:>18.2f}" for n in names))
+ lines.append(f"{'ramping_average':<34}" + "".join(f"{results[n]['ramping_average']:>18.4f}" for n in names))
+ return "\n".join(lines)
+
+
+def main(argv: list[str] | None = None) -> int:
+ dirs = [Path(a) for a in (argv if argv is not None else sys.argv[1:])]
+ if not dirs:
+ print(__doc__.splitlines()[2].strip(), file=sys.stderr)
+ return 2
+ results = {}
+ for d in dirs:
+ env, steps = resimulate(d)
+ results[d.name[-16:]] = split(env, steps)
+ print(table(results))
+ return 0
+
+
+if __name__ == "__main__":
+ raise SystemExit(main())
diff --git a/scripts/tune_strategy.py b/scripts/tune_strategy.py
index d81082e..7a10360 100644
--- a/scripts/tune_strategy.py
+++ b/scripts/tune_strategy.py
@@ -3,6 +3,7 @@
python scripts/tune_strategy.py # version 5 grid (0.7.0)
python scripts/tune_strategy.py --quick # the shipped version 5 defaults only
python scripts/tune_strategy.py --version 4 # the 0.6.0 version 4 search, as it was run
+ python scripts/tune_strategy.py --version 6 # 0.8.0 check: the trim hold, nothing tuned
Tuning set: citylearn_challenge_2023_phase_2_online_evaluation_1 from CityLearn v3.0.2, copied
unchanged into scripts/tuning_data. It has the same three buildings, batteries and LSTM models as
@@ -23,6 +24,12 @@
months, ramping is below the baseline and cost, emissions, discomfort and both peaks are no worse
than the 0.6.0 tuned version 4 strategy on the same months. The qualifying candidate with the
lowest ramping plus daily load factor wins, cost plus emissions breaking ties.
+
+Version 6 (0.8.0) tunes nothing. The envelope's trim hold takes its numbers from the version 5
+strategy (hold within 0.5 C, ease by 0.02 per hour), and the check runs the shipped version 5
+strategy through the 0.7.0 rules and through the 0.8.0 rules (prompt version 6, trim hold on),
+so its effect is measured on the tuning months before anything is reported on the test month.
+The version 4 and 5 searches run with the 0.7.0 rules they were made with.
"""
from __future__ import annotations
@@ -100,8 +107,8 @@ def month_datasets(out: Path) -> list[Path]:
def kpis(job: tuple[str, dict | None, str]) -> dict:
- """job: (kind, params or None, schema path). kind is baseline, rule, strategy (version 4) or
- strategy5 (version 5)."""
+ """job: (kind, params or None, schema path). kind is baseline, rule, strategy (version 4),
+ strategy5 (version 5) or strategy6 (version 5's strategy with the 0.8.0 trim hold)."""
import torch
torch.set_num_threads(1) # one simulation per process
@@ -114,12 +121,16 @@ def kpis(job: tuple[str, dict | None, str]) -> dict:
controller = AgentController(ScriptedSupervisor(), Envelope(EnvelopeConfig.v040()), Budget(1.0), prompt_version=3)
control = True
elif kind == "strategy":
- controller = AgentController(ScriptedSupervisor(strategy=StrategyParams(**params)), Envelope(EnvelopeConfig()),
- Budget(1.0), prompt_version=4)
+ controller = AgentController(ScriptedSupervisor(strategy=StrategyParams(**params)),
+ Envelope(EnvelopeConfig.v070()), Budget(1.0), prompt_version=4)
+ control = True
+ elif kind == "strategy5":
+ controller = AgentController(ScriptedSupervisor(strategy_v5=StrategyParamsV5(**params)),
+ Envelope(EnvelopeConfig.v070()), Budget(1.0), prompt_version=5)
control = True
- else:
- controller = AgentController(ScriptedSupervisor(strategy_v5=StrategyParamsV5(**params)), Envelope(EnvelopeConfig()),
- Budget(1.0), prompt_version=5)
+ else: # strategy6: the same strategy, prompt version 6, with the 0.8.0 trim hold
+ controller = AgentController(ScriptedSupervisor(strategy_v5=StrategyParamsV5(**params)),
+ Envelope(EnvelopeConfig()), Budget(1.0), prompt_version=6)
control = True
res = run_session(CityLearnSimulator(control=control, schema=Path(schema)), controller, NullWriter(),
log=lambda *_: None)
@@ -166,7 +177,7 @@ def line(name: str, k: dict) -> str:
def main(argv: list[str] | None = None) -> int:
ap = argparse.ArgumentParser(description=__doc__.splitlines()[0])
- ap.add_argument("--version", type=int, default=5, choices=(4, 5))
+ ap.add_argument("--version", type=int, default=5, choices=(4, 5, 6))
ap.add_argument("--quick", action="store_true")
ap.add_argument("--jobs", type=int, default=None, help="parallel simulations (default: one per CPU)")
ap.add_argument("--out", type=Path, default=None, help="optionally write every result as JSON")
@@ -174,6 +185,8 @@ def main(argv: list[str] | None = None) -> int:
if args.version == 5:
return main_v5(args)
+ if args.version == 6:
+ return main_v6(args)
with tempfile.TemporaryDirectory() as tmp:
months = [str(d / "schema.json") for d in month_datasets(Path(tmp))]
candidates = [asdict(StrategyParams())] if args.quick else [
@@ -237,5 +250,33 @@ def main_v5(args) -> int:
return 0
+def main_v6(args) -> int:
+ """The trim hold on the tuning months: the shipped version 5 strategy with and without it."""
+ params = asdict(StrategyParamsV5())
+ with tempfile.TemporaryDirectory() as tmp:
+ months = [str(d / "schema.json") for d in month_datasets(Path(tmp))]
+ jobs = [("baseline", None, m) for m in months] + [("strategy5", params, m) for m in months]
+ jobs += [("strategy6", params, m) for m in months]
+ with cf.ProcessPoolExecutor(max_workers=args.jobs, mp_context=mp.get_context("spawn")) as pool:
+ results = list(pool.map(kpis, jobs))
+ n = len(months)
+ base, v5, v6 = mean(results[:n]), mean(results[n:2 * n]), mean(results[2 * n:3 * n])
+ print(f"tuning set: {TUNING_DATASET.name}, {n} months, mean over months")
+ print(line("baseline", base))
+ print(line("0.7 rules", v5))
+ print(line("0.8 hold", v6))
+ worse = [k for k in KEYS if v6[k] > v5[k] + TOLERANCE]
+ print("with the trim hold, worse than without on: " + (", ".join(worse) if worse else "nothing"))
+ for i in range(n):
+ a, b = results[n + i], results[2 * n + i]
+ print(f" month {i + 1}: ramping {a['ramping_average']:.4f} -> {b['ramping_average']:.4f}, "
+ f"cost {a['cost_total']:.4f} -> {b['cost_total']:.4f}, "
+ f"discomfort {a['discomfort_proportion']:.4f} -> {b['discomfort_proportion']:.4f}")
+ if args.out:
+ args.out.write_text(json.dumps({"baseline": base, "v5_070": v5, "v5_080": v6, "months": results},
+ indent=2, sort_keys=True), encoding="utf-8")
+ return 0
+
+
if __name__ == "__main__":
raise SystemExit(main())
diff --git a/tests/test_dryrun.py b/tests/test_dryrun.py
index 8599b31..fc4e7c2 100644
--- a/tests/test_dryrun.py
+++ b/tests/test_dryrun.py
@@ -55,7 +55,7 @@ def call(self, step, request):
def test_cooling_actions_change_the_building():
"""Would have caught a dead HVAC path: a cooling offset through the envelope moves temperatures."""
_, base = _run(BaselineController(), False)
- env = Envelope(EnvelopeConfig(overcool_guard_unoccupied=False))
+ env = Envelope(EnvelopeConfig(overcool_guard_unoccupied=False, trim_hold=False))
_, more = _run(AgentController(ExtraCooling(0.15), env, Budget(1.0)), True)
_, less = _run(AgentController(ExtraCooling(-0.15), env, Budget(1.0)), True)
sent = [s for s in more if s["state"]["baseline"]["Building_1"]["cooling"] < 0.8]
@@ -133,7 +133,7 @@ def test_indoor_temperature_is_the_one_citylearn_simulated():
comfort rule never saw what the heat pump actually did."""
import pandas as pd
- env = Envelope(EnvelopeConfig(overcool_guard_unoccupied=False, cooling_max_deviation=None))
+ env = Envelope(EnvelopeConfig(overcool_guard_unoccupied=False, cooling_max_deviation=None, trim_hold=False))
_, less = _run(AgentController(ExtraCooling(-0.3), env, Budget(1.0)), True, hours=47)
for b in ("Building_1", "Building_2", "Building_3"):
dataset = pd.read_csv(dc.dataset_dir() / f"{b}.csv")["indoor_dry_bulb_temperature"]
@@ -183,3 +183,43 @@ def test_v5_strategy_full_month_brings_ramping_below_the_baseline():
assert k["ramping_average"] < 1.0
assert k["cost_total"] < 0.9611 and k["carbon_emissions_total"] < 0.9947 and k["discomfort_proportion"] < 0.1714
assert k["daily_peak_average"] < 1.0 and k["all_time_peak_average"] < 1.0
+
+
+@needs_citylearn
+def test_v6_dry_run_with_the_trim_hold_and_replay(tmp_path):
+ """24 hours of prompt version 6 (the version 5 strategy, the trim hold on), and replay."""
+ from hyperbolic.sim import CityLearnSimulator
+
+ version, cfg = controller_settings(rule_based=True, prompt_version=6)
+ assert cfg.trim_hold is True
+ ctl = AgentController(ScriptedSupervisor(), Envelope(cfg), Budget(2.0), prompt_version=version)
+ run_dir = tmp_path / "v6"
+ res = run_session(CityLearnSimulator(control=True), ctl, RunWriter(run_dir), hours=HOURS, log=quiet,
+ sim_info={"kind": "citylearn", "control": True})
+ assert res["steps"] == HOURS and res["fallback_steps"] == 0
+ manifest = json.loads((run_dir / "manifest.json").read_text(encoding="utf-8"))
+ assert manifest["supervisor"]["prompt_version"] == 6 and manifest["envelope"]["trim_hold"] is True
+ calls = load_jsonl(run_dir / "calls.jsonl")
+ obs = observation_from_request(calls[5]["request"])
+ assert {"strategy_battery", "cooling_trim_held"} <= set(obs["buildings"]["Building_1"])
+ assert "reference_action" not in json.dumps(obs) and "end_trim" in calls[5]["request"]["system"]
+ rep = replay_run(run_dir)
+ assert rep.deterministic and rep.api_calls == 0 and rep.identical_steps == HOURS
+
+
+@needs_citylearn
+def test_v6_strategy_full_month_with_the_trim_hold():
+ """The scripted version 5 strategy through the 0.8.0 rules: ramping still below the baseline,
+ still ahead of the 0.4.0 rule-based controller, and close to its 0.7.0 month on every KPI."""
+ from hyperbolic.sim import CityLearnSimulator
+
+ version, cfg = controller_settings(rule_based=True, prompt_version=6)
+ ctl = AgentController(ScriptedSupervisor(), Envelope(cfg), Budget(2.0), prompt_version=version)
+ w = NullWriter()
+ res = run_session(CityLearnSimulator(control=True), ctl, w, hours=dc.EPISODE_HOURS, log=quiet)
+ k = res["district_kpis"]
+ assert res["complete"] and res["fallback_steps"] == 0
+ assert k["ramping_average"] < 0.9966 # the version 5 month under the 0.7.0 rules
+ assert k["cost_total"] < 0.9611 and k["carbon_emissions_total"] < 0.9947 and k["discomfort_proportion"] < 0.1714
+ assert abs(k["cost_total"] - 0.9015) < 0.001 and abs(k["discomfort_proportion"] - 0.0424) < 0.002
+ assert any(h["action"] == "hold" for s in w.steps for h in s["decision"].get("holds", []))
diff --git a/tests/test_envelope.py b/tests/test_envelope.py
index 283af8e..e84ac3d 100644
--- a/tests/test_envelope.py
+++ b/tests/test_envelope.py
@@ -191,7 +191,8 @@ def test_config_is_respected():
def test_cooling_deviation_cap(envelope_v3):
- d = envelope_v3.apply(proposal(Building_1={"cooling": 0.7}, Building_2={"cooling": 0.0}), ctx())
+ d = envelope_v3.apply(proposal(Building_1={"cooling": 0.7}, Building_2={"cooling": 0.0}),
+ ctx(Building_2={"indoor": 23.0})) # a cold room, where a trim may run
assert ("Building_1.cooling", "deviation") in rules(d)
assert ("Building_2.cooling", "deviation") in rules(d)
assert d.actions["Building_1"]["cooling"] == pytest.approx(0.55)
diff --git a/tests/test_evaluation.py b/tests/test_evaluation.py
new file mode 100644
index 0000000..89c5ff1
--- /dev/null
+++ b/tests/test_evaluation.py
@@ -0,0 +1,50 @@
+"""scripts/split_ramping.py and scripts/evaluate_hold.py: the 0.8.0 diagnosis and estimate tools."""
+
+import dataclasses
+import json
+import sys
+from pathlib import Path
+
+import pytest
+from conftest import needs_citylearn
+
+sys.path.insert(0, str(Path(__file__).parents[1] / "scripts"))
+
+import evaluate_hold # noqa: E402
+import split_ramping # noqa: E402
+
+from hyperbolic.envelope import Envelope, EnvelopeConfig # noqa: E402
+from hyperbolic.model import Budget, ScriptedSupervisor # noqa: E402
+from hyperbolic.runner import AgentController, RunWriter, run_session # noqa: E402
+
+
+@pytest.mark.parametrize("start, deltas", [(0.5, [0.3, -0.9, 0.2]), (-1.0, [0.4, 0.4, 0.4]), (2.0, [0.0, -0.5])])
+def test_shapley_shares_add_up_to_the_rise(start, deltas):
+ phi = split_ramping._shapley(start, deltas)
+ assert sum(phi) == pytest.approx(max(0.0, start + sum(deltas)) - max(0.0, start))
+
+
+def test_equal_causes_get_equal_shares():
+ a, b = split_ramping._shapley(0.0, [0.3, 0.3])
+ assert a == pytest.approx(b) == pytest.approx(0.3)
+
+
+@needs_citylearn
+def test_evaluation_control_reproduces_a_recorded_run_and_the_split_adds_up(tmp_path):
+ from hyperbolic.sim import CityLearnSimulator
+
+ run = tmp_path / "v5"
+ ctl = AgentController(ScriptedSupervisor(), Envelope(EnvelopeConfig.v070()), Budget(2.0), prompt_version=5)
+ run_session(CityLearnSimulator(control=True), ctl, RunWriter(run), hours=47, log=lambda *_: None,
+ sim_info={"kind": "citylearn", "control": True})
+ published = json.loads((run / "result.json").read_text(encoding="utf-8"))["district_kpis"]
+ control = evaluate_hold.evaluate(run, EnvelopeConfig.v070())
+ assert control["result"]["district_kpis"] == published
+ s = control["split"]
+ assert s["ramping_average"] == pytest.approx(published["ramping_average"], abs=1e-6)
+ assert s["reference_kwh"] + sum(s["shares_kwh"].values()) == pytest.approx(s["ramping_kwh"])
+ hold = evaluate_hold.evaluate(run, dataclasses.replace(EnvelopeConfig.v070(), trim_hold=True))
+ assert hold["result"]["steps"] == 47
+ env, steps = split_ramping.resimulate(run)
+ assert split_ramping.split(env, steps)["ramping_kwh"] == pytest.approx(s["ramping_kwh"])
+ assert evaluate_hold.trim_episodes(control["steps"])["episodes"] >= 0
diff --git a/tests/test_prompt.py b/tests/test_prompt.py
index 0fd1184..e33ce08 100644
--- a/tests/test_prompt.py
+++ b/tests/test_prompt.py
@@ -31,15 +31,20 @@ def fake_states(n):
return out
-def test_default_prompt_is_version_5_and_the_rule_based_controller_stays_on_3():
- assert PROMPT_VERSION == 5
- assert AgentController(ScriptedSupervisor(), Envelope(), Budget(1.0)).describe()["prompt_version"] == 5
+def test_default_prompt_is_version_6_and_the_rule_based_controller_stays_on_3():
+ assert PROMPT_VERSION == 6
+ assert AgentController(ScriptedSupervisor(), Envelope(), Budget(1.0)).describe()["prompt_version"] == 6
# The rule-based controller is the frozen 0.4.0 reference: prompt version 3, no spread check.
assert controller_settings(True) == (3, EnvelopeConfig.v040())
- assert controller_settings(False) == (5, EnvelopeConfig())
- assert controller_settings(True, 4) == (4, EnvelopeConfig())
- assert controller_settings(True, 5) == (5, EnvelopeConfig())
+ assert controller_settings(False) == (6, EnvelopeConfig())
+ # Versions before 6 keep the 0.7.0 rules; only version 6 knows the envelope holds trims.
+ assert controller_settings(True, 4) == (4, EnvelopeConfig.v070())
+ assert controller_settings(True, 5) == (5, EnvelopeConfig.v070())
+ assert controller_settings(False, 5) == (5, EnvelopeConfig.v070())
+ assert controller_settings(True, 6) == (6, EnvelopeConfig())
assert EnvelopeConfig.v040().spread_check is False and EnvelopeConfig().spread_check is True
+ assert EnvelopeConfig().trim_hold is True
+ assert EnvelopeConfig.v070().trim_hold is False and EnvelopeConfig.v040().trim_hold is False
def test_lookahead_places_forecasts_at_their_real_horizons():
@@ -118,15 +123,15 @@ def test_v2_prompt_states_the_strategy_and_the_rules():
def test_request_carries_the_observation_for_each_version():
st = fake_states(1)[0]
ctx = context_from_state(st)
- for version in (1, 2, 3, 4, 5):
+ for version in (1, 2, 3, 4, 5, 6):
obs = observe(version, st, ctx, None, Lookahead())
req = build_request(obs, EnvelopeConfig(), tuple(st["buildings"]), "haiku", prompt_version=version)
assert observation_from_request(req) == json.loads(json.dumps(obs))
with pytest.raises(ValueError):
- observe(6, st, ctx, None, Lookahead())
+ observe(7, st, ctx, None, Lookahead())
-@pytest.mark.parametrize("version", [3, 4, 5])
+@pytest.mark.parametrize("version", [3, 4, 5, 6])
def test_rule_policy_charges_cheap_and_discharges_into_the_peak(version):
# Fake district: price 0.05 most of the day and 0.25 from hour 16 to 20. Version 3 runs the
# 0.4.0 rule; versions 4 and 5 run their scripted strategies, without peak shaving, which the
diff --git a/tests/test_prompt_v6.py b/tests/test_prompt_v6.py
new file mode 100644
index 0000000..f534993
--- /dev/null
+++ b/tests/test_prompt_v6.py
@@ -0,0 +1,163 @@
+"""Prompt version 6 (0.8.0): the trim hold in the prompt, end_trim in the reply, and the version 5
+strategy's battery plan in place of the 0.4.0 reference action."""
+
+import json
+
+import pytest
+from fakes import FakeDistrict
+
+from hyperbolic import district as dc
+from hyperbolic.envelope import Envelope, EnvelopeConfig, parse_reply, proposal_json_schema
+from hyperbolic.model import (
+ Budget,
+ RecordedReplySupervisor,
+ ScriptedSupervisor,
+ battery_plan_v5,
+ observation_from_request,
+ strategy_plan_v5,
+)
+from hyperbolic.prompt import Lookahead, build_request, observe, system_prompt_v5, system_prompt_v6
+from hyperbolic.runner import AgentController, NullWriter, RunWriter, context_from_state, run_session
+from hyperbolic.signals import PeakTracker, PriceOutlook
+
+BUILDINGS = ("Building_1", "Building_2", "Building_3")
+
+
+def _obs(version, n=40, envelope=None):
+ """An observation late in the second day of the fake district, so every outlook is filled."""
+ sim = FakeDistrict()
+ st = sim.start(n + 3)
+ look, peaks = Lookahead(), PeakTracker()
+ for _ in range(n):
+ look.update(st)
+ peaks.update(st)
+ last = st
+ _, st = sim.step(None)
+ t = last["time_step"]
+ ctx = context_from_state(last, peaks.info(t), PriceOutlook(last["shared"]["price_forecast"][0],
+ tuple(look.outlook(t)["price"])))
+ env = envelope or Envelope(EnvelopeConfig())
+ return last, ctx, look, env, observe(version, last, ctx, None, look, env)
+
+
+def test_schema_adds_a_nullable_end_trim_per_building_only_for_version_6():
+ v5, v6 = proposal_json_schema(BUILDINGS), proposal_json_schema(BUILDINGS, end_trim=True)
+ assert "end_trim" not in v5["properties"]["Building_1"]["properties"]
+ dev = v6["properties"]["Building_1"]
+ assert dev["properties"]["end_trim"] == {"anyOf": [{"type": "string"}, {"type": "null"}]}
+ assert "end_trim" in dev["required"] and dev["additionalProperties"] is False
+ assert proposal_json_schema(BUILDINGS) == v5 # the cached older schema is not touched
+
+
+def _reply(**b1):
+ r = {"reason": "t"}
+ for b in BUILDINGS:
+ r[b] = {"cooling": None, "dhw": None, "battery": None, **(b1 if b == "Building_1" else {})}
+ return json.dumps(r)
+
+
+def test_end_trim_parses_for_version_6_and_is_refused_before_it():
+ p, err = parse_reply(_reply(end_trim="room needs it"), BUILDINGS, end_trim=True)
+ assert err is None and p["end_trim"] == {"Building_1": "room needs it"}
+ assert set(p["buildings"]["Building_1"]) == {"cooling", "dhw", "battery"}
+ p, err = parse_reply(_reply(end_trim=None), BUILDINGS, end_trim=True)
+ assert err is None and p["end_trim"] == {}
+ p, err = parse_reply(_reply(), BUILDINGS, end_trim=True) # left out: same as null
+ assert err is None and p["end_trim"] == {}
+ for bad in ("", "x" * (dc.REASON_MAX_CHARS + 1), 3):
+ p, err = parse_reply(_reply(end_trim=bad), BUILDINGS, end_trim=True)
+ assert p is None and "end_trim" in err
+ p, err = parse_reply(_reply(end_trim="no"), BUILDINGS) # version 5 has no such field
+ assert p is None and "end_trim" in err
+ p, err = parse_reply(_reply(), BUILDINGS)
+ assert err is None and "end_trim" not in p
+
+
+def test_v6_observation_swaps_the_reference_action_for_the_strategy_battery_plan():
+ env = Envelope(EnvelopeConfig())
+ st, ctx, look, env, v6 = _obs(6, envelope=env)
+ v5 = observe(5, st, ctx, None, look, env)
+ assert "reference_reason" not in v6 and "reference_action" not in json.dumps(v6)
+ assert {k for k in v6 if k != "strategy_battery_reason"} == {k for k in v5 if k != "reference_reason"}
+ battery, why = battery_plan_v5(v5)
+ assert v6["strategy_battery_reason"] == why
+ for b in BUILDINGS:
+ assert set(v6["buildings"][b]) == set(v5["buildings"][b]) - {"reference_action"} | {
+ "strategy_battery", "cooling_trim_held"}
+ assert v6["buildings"][b]["strategy_battery"] == battery[b]
+ assert v6["buildings"][b]["cooling_trim_held"] == 0.0
+
+
+def test_v6_observation_shows_the_trim_the_envelope_will_hold():
+ env = Envelope(EnvelopeConfig())
+ env.holds = {"Building_2": 0.06}
+ st, ctx, look, env, obs = _obs(6, envelope=env)
+ st_b2 = ctx.buildings["Building_2"]
+ assert obs["buildings"]["Building_2"]["cooling_trim_held"] == -round(env.held_trim("Building_2", st_b2)[0], 3)
+
+
+def test_battery_plan_is_the_v5_strategys_own_battery_action():
+ st, ctx, look, env, v5 = _obs(5)
+ battery, why = battery_plan_v5(v5)
+ full = strategy_plan_v5(v5)
+ assert battery == {b: full[b]["battery"] for b in BUILDINGS}
+ assert why and len(why) <= dc.REASON_TARGET_CHARS and "cooling" not in why
+
+
+def test_v6_system_prompt_states_the_hold_and_the_early_end():
+ cfg = EnvelopeConfig()
+ text = system_prompt_v6(cfg, BUILDINGS)
+ for phrase in ("holds a trim in code", "more than 0.5 C below its setpoint", "at most 0.02 per hour",
+ "end_trim", "strategy_battery", "null keeps a held trim", "cooling above a held trim"):
+ assert phrase in text
+ assert "reference_action" not in text and chr(0x2014) not in text
+ assert "held trim" not in system_prompt_v6(EnvelopeConfig.v070(), BUILDINGS).split("It will clamp:")[1]
+
+
+def test_v6_request_stays_close_to_v5_in_size():
+ st, ctx, look, env, v6 = _obs(6)
+ v5 = observe(5, st, ctx, None, look, env)
+ r5 = build_request(v5, env.config, BUILDINGS, "haiku", prompt_version=5)
+ r6 = build_request(v6, env.config, BUILDINGS, "haiku", prompt_version=6)
+ assert "end_trim" in json.dumps(r6["output_config"]) and "end_trim" not in json.dumps(r5["output_config"])
+ assert len(json.dumps(r6)) < 1.06 * len(json.dumps(r5))
+ assert len(r6["system"]) < 1.07 * len(system_prompt_v5(env.config, BUILDINGS))
+
+
+def test_scripted_v6_reply_follows_the_strategy_battery_and_never_ends_a_trim():
+ st, ctx, look, env, obs = _obs(6)
+ req = build_request(obs, env.config, BUILDINGS, "haiku", prompt_version=6)
+ res = ScriptedSupervisor().call(5, req)
+ reply = json.loads(res.text)
+ for b in BUILDINGS:
+ assert reply[b]["battery"] == obs["buildings"][b]["strategy_battery"] and reply[b]["end_trim"] is None
+ p, err = parse_reply(res.text, BUILDINGS, end_trim=True)
+ assert err is None and p["end_trim"] == {}
+
+
+def test_v6_run_on_the_fake_district_is_clean():
+ ctl = AgentController(ScriptedSupervisor(), Envelope(EnvelopeConfig()), Budget(2.0), prompt_version=6)
+ w = NullWriter()
+ res = run_session(FakeDistrict(), ctl, w, hours=48, log=lambda *_: None)
+ assert res["fallback_steps"] == 0 and ctl.describe()["prompt_version"] == 6
+
+
+def test_recorded_replies_are_served_by_hour_whatever_the_request(tmp_path):
+ run = tmp_path / "v6"
+ ctl = AgentController(ScriptedSupervisor(tokens=(5000, 780)), Envelope(EnvelopeConfig()), Budget(2.0),
+ prompt_version=6)
+ run_session(FakeDistrict(), ctl, RunWriter(run), hours=24, log=lambda *_: None)
+ sup = RecordedReplySupervisor(run / "calls.jsonl", "haiku")
+ calls = [json.loads(x) for x in (run / "calls.jsonl").read_text(encoding="utf-8").splitlines()]
+ got = sup.call(3, {"messages": [{"role": "user", "content": "anything"}]})
+ assert got.text == calls[3]["reply"]["text"] and got.input_tokens == 5000
+ assert sup.call(99, {}).error == "no recorded reply for this hour"
+ assert observation_from_request(calls[3]["request"])["time_step"] == 3
+
+
+@pytest.mark.parametrize("version", [1, 2, 3, 4, 5])
+def test_older_prompt_versions_build_the_same_requests(version):
+ """Requests for versions 1 to 5 carry no end_trim, so recorded runs keep their hashes."""
+ st, ctx, look, env, obs = _obs(version)
+ req = build_request(obs, EnvelopeConfig.v070(), BUILDINGS, "haiku", prompt_version=version)
+ assert "end_trim" not in json.dumps(req)
diff --git a/tests/test_smoothing.py b/tests/test_smoothing.py
index db8a74c..36ccf42 100644
--- a/tests/test_smoothing.py
+++ b/tests/test_smoothing.py
@@ -51,7 +51,7 @@ def test_peak_tracker_reports_last_hour_and_next_hour_as_yesterday():
def test_envelope_remembers_each_buildings_trim_and_forgets_it_on_fallback():
st, ctx, *_ = _obs()
- env = Envelope(EnvelopeConfig())
+ env = Envelope(EnvelopeConfig.v070()) # the record alone; the fake room sits at its setpoint
base = st["baseline"]["Building_1"]["cooling"]
prop = {"reason": "t", "buildings": {b: {"cooling": None, "dhw": None, "battery": None} for b in st["buildings"]}}
prop["buildings"]["Building_1"]["cooling"] = base - 0.08
diff --git a/tests/test_trim_hold.py b/tests/test_trim_hold.py
new file mode 100644
index 0000000..28c9664
--- /dev/null
+++ b/tests/test_trim_hold.py
@@ -0,0 +1,207 @@
+"""The envelope's cooling trim hold (0.8.0): on, hold, ease, end, early end, logging and replay."""
+
+import json
+
+import pytest
+from conftest import BUILDINGS, bstate, proposal
+from fakes import FakeDistrict
+
+from hyperbolic.envelope import Envelope, EnvelopeConfig, StepContext, parse_reply
+from hyperbolic.model import Budget, ScriptedSupervisor
+from hyperbolic.runner import AgentController, RunWriter, load_jsonl, replay_run, run_session
+
+COLD, NEAR, AT = 22.5, 23.7, 24.0 # indoor C against a 24.0 C setpoint: 1.5 below, 0.3 below, at it
+
+
+def step(t=10, **b1) -> StepContext:
+ """Building_1 as given (indoor, occupied, cooling_base); the others at their setpoint."""
+ return StepContext(time_step=t, buildings={b: bstate(**(b1 if b == "Building_1" else {})) for b in BUILDINGS})
+
+
+def hold_rules(d):
+ return [c for c in d.clamps if c.rule == "trim_hold"]
+
+
+def actions(d):
+ return [h["action"] for h in d.holds]
+
+
+@pytest.fixture
+def env():
+ """A trim of 0.08 (cooling 0.32 against a 0.4 baseline) already started in a cold room."""
+ e = Envelope(EnvelopeConfig())
+ d = e.apply(proposal(Building_1={"cooling": 0.32}), step(indoor=COLD))
+ assert actions(d) == ["start"] and not d.clamps and e.holds == {"Building_1": 0.08}
+ return e
+
+
+def test_a_trim_starts_and_is_recorded(env):
+ assert env.trims["Building_1"] == pytest.approx(-0.08)
+
+
+def test_hold_keeps_the_trim_while_the_room_is_more_than_half_a_degree_cold(env):
+ d = env.apply(proposal(), step(indoor=22.8)) # null cooling: the held trim stays on
+ assert d.actions["Building_1"]["cooling"] == pytest.approx(0.32)
+ assert actions(d) == ["hold"] and len(hold_rules(d)) == 1
+ d = env.apply(proposal(Building_1={"cooling": 0.47}), step(indoor=22.9, cooling_base=0.5))
+ assert d.actions["Building_1"]["cooling"] == pytest.approx(0.42) # follows the baseline, same depth
+ c = hold_rules(d)[0]
+ assert c.point == "Building_1.cooling" and c.requested == 0.47 and "more than 0.5 C below" in c.reason
+
+
+def test_a_deeper_trim_is_allowed(env):
+ d = env.apply(proposal(Building_1={"cooling": 0.3}), step(indoor=COLD))
+ assert d.actions["Building_1"]["cooling"] == 0.3 and not d.clamps and not d.holds
+ assert env.holds["Building_1"] == pytest.approx(0.1)
+
+
+def test_within_half_a_degree_the_trim_eases_by_at_most_0_02_per_hour(env):
+ d = env.apply(proposal(), step(indoor=NEAR))
+ assert d.actions["Building_1"]["cooling"] == pytest.approx(0.34) and actions(d) == ["ease"]
+ assert "eases off" in hold_rules(d)[0].reason
+ for want in (0.36, 0.38):
+ d = env.apply(proposal(Building_1={"cooling": 0.4}), step(indoor=NEAR))
+ assert d.actions["Building_1"]["cooling"] == pytest.approx(want)
+ d = env.apply(proposal(), step(indoor=NEAR))
+ assert d.actions["Building_1"]["cooling"] is None and actions(d) == ["release"]
+ assert env.holds == {}
+
+
+def test_a_miss_smaller_than_rounding_is_not_clamped_but_still_held(env):
+ d = env.apply(proposal(Building_1={"cooling": 0.3215}), step(indoor=COLD)) # 0.0015 short
+ assert not d.clamps and env.holds["Building_1"] == pytest.approx(0.08)
+ d = env.apply(proposal(Building_1={"cooling": 0.3245}), step(indoor=COLD)) # still measured from 0.08
+ assert d.actions["Building_1"]["cooling"] == pytest.approx(0.32) and len(hold_rules(d)) == 1
+
+
+@pytest.mark.parametrize("where, why", [
+ ({"indoor": AT}, "at or above the setpoint"),
+ ({"indoor": 24.6}, "at or above the setpoint"),
+ ({"indoor": COLD, "occupied": False}, "unoccupied"),
+ ({"indoor": COLD, "cooling_base": 0.0}, "no cooling to trim"),
+])
+def test_the_trim_ends_at_once(env, where, why):
+ d = env.apply(proposal(), step(**where))
+ assert actions(d) == ["end"] and why in d.holds[0]["reason"]
+ assert d.actions["Building_1"]["cooling"] is None and not d.clamps and env.holds == {}
+
+
+def test_a_trim_asked_for_where_the_hold_has_ended_goes_back_to_the_baseline(env):
+ d = env.apply(proposal(Building_1={"cooling": 0.3}), step(indoor=AT))
+ assert d.actions["Building_1"]["cooling"] == 0.4
+ assert "trim ended" in hold_rules(d)[0].reason and env.holds == {}
+ fresh = Envelope(EnvelopeConfig()) # no trim on yet: none may start in an empty room either
+ d = fresh.apply(proposal(Building_1={"cooling": 0.3}), step(indoor=COLD, occupied=False))
+ assert d.actions["Building_1"]["cooling"] == 0.4 and actions(d) == ["end"] and fresh.holds == {}
+
+
+def _v6_reply(**b1) -> dict:
+ reply = {"reason": "test"}
+ for b in BUILDINGS:
+ reply[b] = {"cooling": None, "dhw": None, "battery": None, "end_trim": None, **(b1 if b == "Building_1" else {})}
+ p, err = parse_reply(json.dumps(reply), BUILDINGS, end_trim=True)
+ assert err is None
+ return p
+
+
+def test_the_model_can_end_a_trim_early_only_with_a_reason(env):
+ d = env.apply(_v6_reply(end_trim="Room heading for an outage, keep it cool"), step(indoor=COLD))
+ assert d.actions["Building_1"]["cooling"] is None and not d.clamps
+ assert d.holds == [{"building": "Building_1", "action": "end_early", "depth_before": 0.08, "depth_after": 0.0,
+ "reason": "Room heading for an outage, keep it cool"}]
+ assert env.holds == {}
+
+
+def test_an_early_end_that_still_asks_for_a_trim_is_brought_back_to_the_baseline(env):
+ d = env.apply(_v6_reply(cooling=0.3, end_trim="done"), step(indoor=COLD))
+ assert d.actions["Building_1"]["cooling"] == 0.4 and actions(d) == ["end_early"]
+ assert "ended it early: done" in hold_rules(d)[0].reason
+
+
+def test_end_trim_without_a_trim_on_does_nothing():
+ e = Envelope(EnvelopeConfig())
+ d = e.apply(_v6_reply(end_trim="nothing to end"), step(indoor=COLD))
+ assert not d.holds and not d.clamps and d.status == "ok"
+
+
+def test_every_hold_action_is_logged_in_the_decision(env):
+ d = env.apply(proposal(), step(indoor=22.8))
+ rec = d.as_dict()
+ assert rec["status"] == "clamped" and rec["clamps"][0]["rule"] == "trim_hold"
+ assert rec["holds"][0]["action"] == "hold" and rec["holds"][0]["depth_after"] == 0.08
+ quiet = env.apply(proposal(Building_1={"cooling": 0.32}), step(indoor=22.8)).as_dict()
+ assert "holds" not in quiet and quiet["status"] == "ok" # nothing to record, no key
+
+
+def test_fallback_forgets_the_hold(env):
+ env.fallback(step(indoor=COLD), "test")
+ assert env.holds == {}
+ d = env.apply(proposal(), step(indoor=COLD))
+ assert d.actions["Building_1"]["cooling"] is None and not d.holds
+
+
+def test_deviation_and_comfort_rules_still_come_first():
+ e = Envelope(EnvelopeConfig())
+ d = e.apply(proposal(Building_1={"cooling": 0.1}), step(indoor=COLD))
+ assert {c.rule for c in d.clamps} == {"deviation"} and e.holds["Building_1"] == pytest.approx(0.15)
+ # Near the bottom of the band extra cooling is still refused, and the trim is untouched.
+ d = e.apply(proposal(Building_1={"cooling": 0.5}), step(indoor=22.2))
+ assert {c.rule for c in d.clamps} == {"comfort", "trim_hold"}
+ assert d.actions["Building_1"]["cooling"] == pytest.approx(0.25)
+
+
+def test_without_the_hold_the_0_7_0_rules_are_unchanged():
+ e = Envelope(EnvelopeConfig.v070())
+ e.apply(proposal(Building_1={"cooling": 0.32}), step(indoor=COLD))
+ d = e.apply(proposal(), step(indoor=22.8))
+ assert d.actions["Building_1"]["cooling"] is None and not d.clamps and not d.holds and e.holds == {}
+ assert "holds" not in d.as_dict()
+
+
+def test_older_runs_rebuild_their_rules_without_the_hold():
+ old = EnvelopeConfig.v070().as_dict()
+ for key in ("trim_hold", "trim_hold_c", "trim_release_step", "trim_min_depth"):
+ old.pop(key)
+ assert EnvelopeConfig.from_manifest(old) == EnvelopeConfig.v070() # a 0.6.0 or 0.7.0 manifest
+ assert EnvelopeConfig.from_manifest({}).trim_hold is False
+ assert EnvelopeConfig.from_manifest(EnvelopeConfig().as_dict()) == EnvelopeConfig()
+
+
+class ColdDistrict(FakeDistrict):
+ """The fake district with a 25 C setpoint, so occupied rooms start a degree too cold and the
+ scripted strategy trims them; trimming warms them (FakeDistrict's indoor model)."""
+
+ def state(self):
+ st = super().state()
+ for s in st["buildings"].values():
+ s["setpoint_c"] = 25.0
+ return st
+
+
+def _record_v6(run_dir):
+ ctl = AgentController(ScriptedSupervisor(tokens=(5000, 780)), Envelope(EnvelopeConfig()), Budget(2.0),
+ prompt_version=6)
+ run_session(ColdDistrict(), ctl, RunWriter(run_dir), hours=48, log=lambda *_: None)
+
+
+def test_a_v6_run_with_hold_actions_replays_byte_for_byte(tmp_path):
+ run = tmp_path / "v6"
+ _record_v6(run)
+ steps = load_jsonl(run / "steps.jsonl")
+ kinds = {h["action"] for s in steps for h in s["decision"].get("holds", [])}
+ assert {"start", "hold", "release"} <= kinds
+ assert any(c["rule"] == "trim_hold" for s in steps for c in s["decision"]["clamps"])
+ assert json.loads((run / "manifest.json").read_text(encoding="utf-8"))["envelope"]["trim_hold"] is True
+ rep = replay_run(run)
+ assert rep.deterministic and rep.api_calls == 0 and rep.identical_steps == 48, (rep.mismatched_steps, rep.error)
+ replay_run(run, out_dir=tmp_path / "again")
+ assert (tmp_path / "again" / "steps.jsonl").read_bytes() == (run / "steps.jsonl").read_bytes()
+
+
+def test_a_run_replayed_under_other_hold_rules_is_detected(tmp_path):
+ run = tmp_path / "v6"
+ _record_v6(run)
+ m = json.loads((run / "manifest.json").read_text(encoding="utf-8"))
+ m["envelope"]["trim_release_step"] = 1.0 # release at once instead
+ (run / "manifest.json").write_text(json.dumps(m), encoding="utf-8")
+ assert not replay_run(run).deterministic
diff --git a/tests/test_viewer.py b/tests/test_viewer.py
index eb913b8..9771756 100644
--- a/tests/test_viewer.py
+++ b/tests/test_viewer.py
@@ -198,3 +198,33 @@ def test_page_explains_how_it_works_and_how_to_run_it():
assert cmd in run, cmd
assert '