WP4: evidence scripts - timer baseline, constrained site, forecast, fit, multi-day

AGENTS.md

@@ -7,7 +7,7 @@ Hackathon entry: "Noonshift" — a CPO-side, deadline-based EV charging schedule
 ## Stack & Commands
 - Python 3.12/3.13 · FastAPI · scipy 1.14 (HiGHS LP) · mobilityhouse/ocpp · asyncpg/Postgres (optional) · Docker Compose. Two Vite front-ends: `web/` (ops dashboard, React+JSX+Tailwind 3, :5173) and `wattwise/` (driver app, React+TS+Tailwind 4, :5174); `npm install && npm run dev` in each, both proxy `/api` and `/ws` to :8000. `npm run build` in each, then `docker compose up` serves them on :3000 / :3001.
 - `python -m venv .venv && .venv/Scripts/pip install -r requirements-dev.txt`
-- `python -m pytest` (69 tests, ~90 s) · `python scripts/prove.py` (slide 1) · `python -m noonshift.test_loop` · `docker compose up -d --build`
+- `python -m pytest` (70 tests, ~90 s) · `python scripts/prove.py` (slide 1) · `python -m noonshift.test_loop` · `docker compose up -d --build`
 - Data: `python -m noonshift.seed gen` (placeholders) · `python scripts/fetch_data.py signal|caiso|sessions --day ...` (real).
 
 ## Current State & Focus
@@ -33,10 +33,12 @@ Signals (WattTime MOER; CAISO fuel-mix fallback) + tariff table + sessions (dead
 - `wattwise/src/api/noonshift.ts` — driver app client + hand-copied types; `connectWs()`, `onSimDay()` (ready-by is built on the **sim** day and must be > sim now).
 - `wattwise/src/context/WattwiseContext.tsx` — driver state: `POST /sessions` on the highest free connector (replayed sessions use c01–c36), `/live` polled every 2 s for the receipt, meter/plan frames for kW/kWh/window; unplug/day_reset moves the session to History.
 - `noonshift/ocpp_gateway.py` — two profiles per connector: static share at stack 0 (sent as a task after BootNotification, no expiry) + dynamic plan at stack 1 with `valid_to` = now + 15 min, refreshed every 10 sim-min; `Charger.profiles`, `Charger.apply(now)`; self-check `python -m noonshift.ocpp_gateway`. Never `await self.call()` inside an `@after` handler (deadlocks the message loop). `db.py`, `seed.py`, `test_loop.py` — Tirth's.
-- `scripts/prove.py` — two real-loop replays (charge-now vs Noonshift), slide-1 table, gate exit code.
+- `scripts/prove.py` — real-loop replays: charge-now vs Noonshift (+ `--policy timer` = dumb 09–14 timer; `--block/--feed` overrides; `--forecast <file>` plans on it and scores on data/signal.json); prints clean-hour share, block overage $, kWh short, stated-vs-actual departures; gate exit code. Timer delta on the real day: CO2 −56 %, bill −28.5 %, peak −28 %. At feed 80/block 50: CO2 −23 % but bill +7 % (deferred energy lands in the 16–21 h price peak).
+- `scripts/fit.py <sessions.json|csv>` — pre-sales site fit: `cash tiers | perks only | not a fit` from stated-slack median, tariff spread, demand charge, `--saving-rate` (measured). Caltech: perks only (slack median 3.8 h, 4/15/17).
+- `scripts/replay_days.py --days a,b,c` — min/median/max over `data/days/<date>.json` signals; names missing days instead of inventing them.
 - `scripts/smoke.py` — e2e without a browser: driver REST (`/price`, `/sessions`, `/boost`, `/live`) + `/ws` as the ops app sees it + all four `/demo/*` + ladder drop/restore; plugs 2 extra cars before the demo beats so early-unplug/boost always have a target.
 - `.docs/pitch/` — `deck.md` (slide list, all sourced), `demo-script.md` (4 min, 7 beats, what the audience sees per beat, failure modes), `hard-questions.md` (answers tied to code + tests).
-- `scripts/fetch_data.py` — WattTime / ACN-Data / CAISO fallback → `data/*.json`.
+- `scripts/fetch_data.py` — WattTime / ACN-Data / CAISO fallback → `data/*.json`; `signal --forecast` → `data/signal_forecast.json` via `/v3/forecast/historical` (untested: no WattTime creds on this machine); `signal` also stores `health_damage` ($/MWh) when the plan serves it → `impact(..., health=True)["health_usd"]`.
 - `tests/test_scheduler.py`, `test_impact.py`, `test_perf.py`, `test_day.py` (full day + demo scenarios), `tests/fixtures/baseline_1405.json`.
 - `.docs/neal-plan.md` — Neal's lane: status vs gates, LP deviations and why, edge-case→test matrix, open items.
 - `.docs/team-plan.md`, `.docs/noonshift-proposal.md`, `ev-green-charging-ideation.html` — plan and evidence base.
@@ -79,6 +81,7 @@ Signals (WattTime MOER; CAISO fuel-mix fallback) + tariff table + sessions (dead
 - 2026-09-12 — Session data = ACN 2019-04-09 re-dated onto the 2026 signal day; kwh_needed = delivered energy, stated departure = the driver's own input (early leavers kept).
 
 ## Changelog
+2026-09-13 | WP4 evidence scripts | scripts/prove.py, fetch_data.py, fit.py, replay_days.py, scheduler.impact(health=) | timer baseline is the judge's number (−56 % CO2 over a dumb timer); constrained site shows the carbon-vs-tariff conflict honestly; forecast flag wired, unverified without creds
 2026-09-13 | WP3 cars, needs, queues | models.py, sim.py, api.py, data/site.json, tests/test_day.py, scripts/smoke.py | vehicle form → per-car cap + kWh from SoC; observation guard believes the meter; connectors_asap c41/c42; done event + idle_min; Sim.waiting + move-by tightening; a notified done car moves within 10 min when someone waits
 2026-09-13 | WP2 fail-safe an inspector accepts: static share under everything | sim.py, api.py, scheduler.py, ocpp_gateway.py, models.py, data/site.json, tests/test_day.py | `full` rung, `apply_limits` fallback and the linprog fallback all return `safe_share_kw`; kill-the-loop test: 6 h with no step() never exceeds headroom and after 15 min never exceeds n × share; `StatusOut.safe_share_kw`
 2026-09-13 | WP1 pricing + emergency + first-hour floor | scheduler.py, api.py, models.py, data/site.json, tests/, scripts/smoke.py, .docs/implementation-plan.md | price() = §7b shared savings (never above R); urgency endpoint; slot-0 remaining-duration + time-anchored checkpoints fixed a lost-energy bug that the 3.5 kWh promise exposed

noonshift/scheduler.py

@@ -253,8 +253,9 @@ def solve(cars, site, signal, tariff, now) -> dict[str, list[float]]:
     return solve_lp(cars, site, signal, tariff, now)[0]
 
 
-def _totals(plan, moer, price, tariff):
-    """kWh, $ (energy + this day's share of block overage), kg CO2 and peak kW of one {cid: [kW per slot]} dict."""
+def _totals(plan, moer, price, tariff, health=None):
+    """kWh, $ (energy + this day's share of block overage), kg CO2, peak kW and (with a health-damage index, $/MWh)
+    health $ of one {cid: [kW per slot]} dict."""
     site_kw = np.zeros(HORIZON)
     for p in plan.values():
         a = np.asarray(list(p)[:HORIZON], dtype=float)
@@ -267,25 +268,34 @@ def _totals(plan, moer, price, tariff):
     block = tariff.get("block_kw")
     if block is not None:
         usd += max(0.0, peak - block) * tariff.get("block_price", 0.0) * tariff.get("overage_multiplier", 2.0) / DAYS_PER_MONTH
-    return {"kwh": kwh, "usd": usd, "kgco2": kg, "peak_kw": peak}
+    out = {"kwh": kwh, "usd": usd, "kgco2": kg, "peak_kw": peak}
+    if health is not None:
+        out["health_usd"] = float((site_kw * SLOT_H * health / 1000).sum())  # $/MWh -> $ per kWh-slot
+    return out
 
 
 def impact_detail(plan, baseline, signal, tariff) -> dict:
     """Both totals plus the savings. Energy-matched: the baseline is scaled to the plan's kWh, so a plan that
     delivered less (shortfall) cannot book the undelivered energy as a saving. No signal => kg CO2 is 0, not guessed."""
     moer, price = _per_slot(signal.get("moer")), _per_slot(tariff.get("price_per_kwh"))
-    p, b = _totals(plan, moer, price, tariff), _totals(baseline, moer, price, tariff)
+    health = _per_slot(signal["health_damage"]) if signal.get("health_damage") else None
+    p, b = _totals(plan, moer, price, tariff, health), _totals(baseline, moer, price, tariff, health)
     ratio = p["kwh"] / b["kwh"] if b["kwh"] > 0 else 0.0
     return {"plan": p, "baseline": b, "energy_matched": abs(p["kwh"] - b["kwh"]) < 1e-6,
             "saved_usd": round(b["usd"] * ratio - p["usd"], 4),
             "saved_kgco2": round(b["kgco2"] * ratio - p["kgco2"], 4) if signal.get("moer") else 0.0,
+            "health_usd": round(b["health_usd"] * ratio - p["health_usd"], 4) if health is not None else None,
             "signal_kind": signal.get("kind")}
 
 
-def impact(plan, baseline, signal, tariff) -> dict[str, float]:
-    """Savings of `plan` vs `baseline` (both {connector_id: [kW per slot]}), keys exactly as LiveOut expects."""
+def impact(plan, baseline, signal, tariff, *, health=False) -> dict[str, float]:
+    """Savings of `plan` vs `baseline` (both {connector_id: [kW per slot]}), keys exactly as LiveOut expects.
+    health=True adds health_usd (metrics.md §4; None when the signal carries no health_damage index)."""
     d = impact_detail(plan, baseline, signal, tariff)
-    return {"saved_usd": d["saved_usd"], "saved_kgco2": d["saved_kgco2"]}
+    out = {"saved_usd": d["saved_usd"], "saved_kgco2": d["saved_kgco2"]}
+    if health:
+        out["health_usd"] = d["health_usd"]
+    return out
 
 
 def tier_of(slack_hours) -> str:

scripts/fetch_data.py

@@ -1,6 +1,7 @@
 """Download the demo day into data/*.json (same schema seed.py writes) so the demo runs offline.
 
-    python scripts/fetch_data.py signal   --day 2026-04-14          # WattTime CAISO_NORTH MOER (needs WATTTIME_USER/PASSWORD)
+    python scripts/fetch_data.py signal   --day 2026-04-14          # WattTime CAISO_NORTH MOER (+ health_damage when the plan returns it)
+    python scripts/fetch_data.py signal   --day 2026-04-14 --forecast   # the MOER forecast as published at 06:00 that day -> data/signal_forecast.json
     python scripts/fetch_data.py caiso    --day 2026-04-14          # fallback: CAISO fuel mix -> AVERAGE intensity, no auth
     python scripts/fetch_data.py sessions --day 2019-04-09 --n 40   # ACN-Data Caltech sessions (needs ACN_TOKEN; data spans 2018-04..2021-09), re-dated to --site-day
 
@@ -51,21 +52,57 @@ def to_slots(points, day):
     return [round(v, 1) for v in out], gaps
 
 
-def watttime(day):
+def _watttime_token():
     user, pw = os.environ.get("WATTTIME_USER"), os.environ.get("WATTTIME_PASSWORD")
     if not (user and pw):
         sys.exit("set WATTTIME_USER and WATTTIME_PASSWORD (free Basic plan covers CAISO_NORTH)")
     auth = base64.b64encode(f"{user}:{pw}".encode()).decode()
-    token = json.loads(get("https://api.watttime.org/login", {"Authorization": f"Basic {auth}"}))["token"]
+    return json.loads(get("https://api.watttime.org/login", {"Authorization": f"Basic {auth}"}))["token"]
+
+
+def _pt(iso):
+    return datetime.fromisoformat(iso.replace("Z", "+00:00")).astimezone(PT)
+
+
+def watttime(day, forecast=False):
+    token = _watttime_token()
     start = datetime.combine(day, datetime.min.time(), PT).astimezone(timezone.utc)
+    hdr = {"Authorization": f"Bearer {token}"}
+    if forecast:
+        # solutions.md §6: the forecast a product would have planned on at 06:00 that morning, not the realised signal.
+        # Assumption: WattTime Basic serves /v3/forecast/historical for CAISO_NORTH; if it does not, run this on a live
+        # day with /v3/forecast and the flag stays wired.
+        at = start + timedelta(hours=6)
+        q = urllib.parse.urlencode({"region": "CAISO_NORTH", "signal_type": "co2_moer", "start": at.isoformat(),
+                                    "end": (at + timedelta(minutes=5)).isoformat(), "horizon_hours": 24})
+        gens = json.loads(get(f"https://api.watttime.org/v3/forecast/historical?{q}", hdr))["data"]
+        if not gens:
+            sys.exit("no historical forecast returned for that day")
+        g = gens[0]
+        points = [(_pt(p["point_time"]), p["value"] * LBS_PER_MWH_TO_G_PER_KWH) for p in g["forecast"]]
+        moer, gaps = to_slots(points, day)
+        json.dump({"kind": "marginal", "region": "CAISO_NORTH", "date": day.isoformat(), "forecast": True,
+                   "generated_at": g.get("generated_at"), "source": "WattTime v3 co2_moer forecast (historical)", "moer": moer},
+                  open("data/signal_forecast.json", "w"))
+        print(f"wrote data/signal_forecast.json: forecast generated {g.get('generated_at')}, {len(points)} points, {gaps} gaps filled "
+              f"(the hours before 06:00 are not in a 06:00 forecast), min {min(moer)} max {max(moer)} g/kWh")
+        return
     q = urllib.parse.urlencode({"region": "CAISO_NORTH", "signal_type": "co2_moer",
                                 "start": start.isoformat(), "end": (start + timedelta(days=1)).isoformat()})
-    data = json.loads(get(f"https://api.watttime.org/v3/historical?{q}", {"Authorization": f"Bearer {token}"}))["data"]
-    points = [(datetime.fromisoformat(p["point_time"].replace("Z", "+00:00")).astimezone(PT), p["value"] * LBS_PER_MWH_TO_G_PER_KWH)
-              for p in data]
+    data = json.loads(get(f"https://api.watttime.org/v3/historical?{q}", hdr))["data"]
+    points = [(_pt(p["point_time"]), p["value"] * LBS_PER_MWH_TO_G_PER_KWH) for p in data]
     moer, gaps = to_slots(points, day)
-    json.dump({"kind": "marginal", "region": "CAISO_NORTH", "date": day.isoformat(), "source": "WattTime v3 co2_moer",
-               "moer": moer}, open("data/signal.json", "w"))
+    out = {"kind": "marginal", "region": "CAISO_NORTH", "date": day.isoformat(), "source": "WattTime v3 co2_moer", "moer": moer}
+    try:  # metrics.md §4: the health-damage index ($/MWh) rides along when the plan serves it
+        q = urllib.parse.urlencode({"region": "CAISO_NORTH", "signal_type": "health_damage",
+                                    "start": start.isoformat(), "end": (start + timedelta(days=1)).isoformat()})
+        hd = json.loads(get(f"https://api.watttime.org/v3/historical?{q}", hdr))["data"]
+        out["health_damage"], hd_gaps = to_slots([(_pt(p["point_time"]), p["value"]) for p in hd], day)
+        out["health_source"] = "WattTime v3 health_damage ($/MWh)"
+        print(f"health_damage: {len(hd)} points, {hd_gaps} gaps filled, mean {sum(out['health_damage']) / SLOTS:.1f} $/MWh")
+    except Exception as e:  # noqa: BLE001
+        print(f"health_damage not stored ({e}); impact.health_usd stays None")
+    json.dump(out, open("data/signal.json", "w"))
     print(f"wrote data/signal.json: WattTime CAISO_NORTH MOER for {day}, {len(points)} points, {gaps} gaps filled, "
           f"min {min(moer)} max {max(moer)} g/kWh")
 
@@ -147,7 +184,8 @@ def sessions(day, n, site_day):
     ap.add_argument("--day", required=True, help="YYYY-MM-DD of the data to fetch")
     ap.add_argument("--site-day", default="2026-04-14", help="sessions: re-date onto this day so all files agree")
     ap.add_argument("--n", type=int, default=40)
+    ap.add_argument("--forecast", action="store_true", help="signal: fetch the 06:00 forecast for --day instead of the realised MOER")
     a = ap.parse_args()
     day = datetime.fromisoformat(a.day).date()
-    {"signal": lambda: watttime(day), "caiso": lambda: caiso(day),
+    {"signal": lambda: watttime(day, a.forecast), "caiso": lambda: caiso(day),
      "sessions": lambda: sessions(day, a.n, datetime.fromisoformat(a.site_day).date())}[a.what]()

scripts/fit.py

@@ -0,0 +1,82 @@
+"""Site-fit check before anyone signs (solutions.md §4, business.md §9b): run it on the customer's own session export.
+
+    python scripts/fit.py data/sessions.json                    # JSON list with arrival, departure, kwh_needed (or kWh)
+    python scripts/fit.py export.csv --p-max 7 --spread 0.04    # CSV with arrival, departure, kwh columns; tariff spread $/kWh
+    python scripts/fit.py export.csv --demand-charge             # the bill has a demand charge (capacity is worth money)
+    python scripts/fit.py data/sessions.json --saving-rate 0.006 # what prove.py measured ($/kWh): below 4 c/kWh, cash tiers are unfunded
+
+Slack per session = dwell - kWh / p_max. Verdict from the median slack and the money on the bill:
+    cash tiers   median slack >= 4 h and (tariff spread >= 4 c/kWh or a demand charge)   -> §7b discounts are real money
+    perks only   median slack >= 1.5 h                                                     -> non-price perks (§9b)
+    not a fit    median slack < 1.5 h                                                      -> nothing to shift
+Same arithmetic as metrics.md §6b "site-type check". Slack uses the driver's stated departure when the export has one
+(that is the slack the scheduler actually gets), else the real one. Prints the numbers it used, so the verdict can be
+argued with. The slack rule is a heuristic: a measured saving rate (prove.py: saved $ / kWh) overrides it.
+"""
+import argparse
+import csv
+import json
+import statistics
+import sys
+from datetime import datetime
+
+
+def load(path):
+    if path.endswith(".json"):
+        rows = json.load(open(path))
+    else:
+        rows = list(csv.DictReader(open(path, newline="")))
+    out = []
+    for r in rows:
+        kwh = next((float(r[k]) for k in ("kwh_needed", "kwh", "kWh", "kWhDelivered", "energy_kwh") if r.get(k) not in (None, "")), None)
+        arr = r.get("arrival") or r.get("connectionTime")
+        dep = r.get("user_stated_departure") or r.get("departure") or r.get("disconnectTime")
+        if kwh is None or not arr or not dep:
+            continue
+        a, d = datetime.fromisoformat(str(arr)), datetime.fromisoformat(str(dep))
+        out.append(((d - a).total_seconds() / 3600, kwh))
+    if not out:
+        sys.exit("no rows with arrival, departure and kWh")
+    return out, any(r.get("user_stated_departure") for r in rows)
+
+
+def verdict(sessions, p_max, spread, demand_charge, saving_rate=None):
+    slack = sorted(dwell - kwh / p_max for dwell, kwh in sessions)
+    med = statistics.median(slack)
+    if med < 1.5:
+        v = "not a fit"
+    elif med >= 4 and (spread >= 0.04 or demand_charge) and (saving_rate is None or saving_rate >= 0.04):
+        v = "cash tiers"
+    else:
+        v = "perks only"
+    return v, med, slack
+
+
+def main(argv=None):
+    ap = argparse.ArgumentParser()
+    ap.add_argument("path")
+    ap.add_argument("--p-max", type=float, default=7.0, help="kW per port")
+    ap.add_argument("--spread", type=float, help="tariff spread $/kWh (max - min); default: from data/tariff.json")
+    ap.add_argument("--demand-charge", action="store_true", help="the site's bill carries a demand charge")
+    ap.add_argument("--saving-rate", type=float, help="measured saving $/kWh from prove.py (saved $ / kWh); overrides the slack rule")
+    a = ap.parse_args(argv)
+    if a.spread is None:
+        try:
+            prices = json.load(open("data/tariff.json"))["price_per_kwh"]
+            a.spread = max(prices) - min(prices)
+        except Exception:  # noqa: BLE001
+            a.spread = 0.0
+    sessions, stated = load(a.path)
+    v, med, slack = verdict(sessions, a.p_max, a.spread, a.demand_charge, a.saving_rate)
+    n = len(slack)
+    print(f"{a.path}: {n} sessions, {'stated' if stated else 'actual'} dwell median {statistics.median(d for d, _ in sessions):.1f} h, "
+          f"kWh median {statistics.median(k for _, k in sessions):.1f}, slack median {med:.1f} h "
+          f"(< 1 h: {sum(s < 1 for s in slack)}, 1-4 h: {sum(1 <= s < 4 for s in slack)}, >= 4 h: {sum(s >= 4 for s in slack)}); "
+          f"tariff spread {100 * a.spread:.1f} c/kWh; demand charge {'yes' if a.demand_charge else 'no'}"
+          + (f"; measured saving {100 * a.saving_rate:.1f} c/kWh" if a.saving_rate is not None else "; measured saving: not given (run prove.py)"))
+    print(f"verdict: {v}")
+    return v
+
+
+if __name__ == "__main__":
+    main()