fix: degradation watch regresses on the whole lookback, not the newest 40 sessions

contrib/backtest_forecast.py

@@ -499,6 +499,7 @@ def main(argv: list[str] | None = None) -> int:
     now = time.time()
     fits = thermal.fit_sessions(db, now, lookback_days=args.lookback_days)
     params = thermal.fit_history(db, now, fits=fits)
+    fits = thermal.forecast_fits(fits)  # what the deployed forecast learns from
     idle_model = thermal.load_idle_offset(db)
     sessions = [
         sess for sess in db.sessions_range(now - args.lookback_days * 86400, now)

docs/thermal-model.md

@@ -285,6 +285,11 @@ row. More sessions either confirm it or dissolve it.
 
 ### What it compares
 
+- **Every session in the lookback.** The forecast learns from the newest
+  40 sessions; the watch does not. A count window makes the baseline
+  whatever happened ~40 charges ago. On one install a single odd week at
+  that edge read as +8 °C of "confirmed" drift, while the 80 days since
+  the verified anchor were flat (−0.04 °C).
 - **Only free-running fits.** Windows the charger was regulating are
   excluded outright — their plateau is a setpoint, not an equilibrium.
   `regulated_n` says how many sat out, and the Alerts page says so too,

tests/test_wallmonitor.py

@@ -2,13 +2,15 @@
 
 import asyncio
 import math
+import threading
 import time
 from statistics import median
 
 import aiohttp
 import pytest
 
 from wallmonitor import thermal
+from wallmonitor import web as web_module
 from wallmonitor.config import Config
 from wallmonitor.db import Database
 from wallmonitor.poller import EventBus, Poller
@@ -1300,6 +1302,29 @@ async def test_thermal_drift_ambient_confound_caps_verdict_at_lead(db):
     assert drift["drifting"] is False and drift["lead"] is True
 
 
+async def test_thermal_drift_baseline_is_whole_lookback_not_newest_40(db):
+    # Regression: fit_sessions once kept only the newest 40 sessions, so the
+    # watch's baseline was whatever happened ~40 charges ago. A flat install
+    # with one low stretch sitting at that edge read as confirmed drift
+    # (seen live: +8 C over 34 days; flat over the 80 days since the anchor).
+    now = time.time()
+    rises = [36.0] * 12 + [29.0] * 6 + [36.0] * 34  # the dip lands at the 40th-newest
+    for i, rise in enumerate(rises):
+        _seed_thermal_session(db, now - (len(rises) - i) * 7200, ambient_c=25.0,
+                              rise_ref_c=rise + (0.3 if i % 2 else -0.3))
+    fits = thermal.fit_sessions(db, now)
+    assert len(fits) == len(rises), "every session in the lookback is fitted"
+    # The forecast still learns from the newest FORECAST_SESSIONS only.
+    params = thermal.fit_history(db, now, fits=fits)
+    assert params.tau_fits == thermal.FORECAST_SESSIONS
+    assert thermal.forecast_fits(fits) == fits[-thermal.FORECAST_SESSIONS:]
+    drift = thermal.detect_drift(fits)
+    assert drift is not None and drift["drifting"] is False
+    assert abs(drift["delta_c"]) < thermal.DRIFT_WARN_C
+    # The window the cap used to impose does alarm on the same data.
+    assert thermal.detect_drift(fits[-40:])["drifting"] is True
+
+
 async def test_thermal_baseline_anchor(db):
     now = time.time()
     rises = [36.0, 36.5, 35.8, 36.2, 42.0, 41.5, 42.3]
@@ -1649,6 +1674,35 @@ async def test_thermal_api_endpoint(db):
         await client.close()
 
 
+async def test_thermal_api_serves_stale_fit_while_refitting(db, monkeypatch):
+    # The history fit can take tens of seconds and the amp controller polls
+    # with a 10 s timeout: a stale fit is served at once while one refit
+    # runs in the background, instead of every poll waiting on it.
+    app = make_app(db, EventBus(), None)
+    async with TestClient(TestServer(app)) as client:
+        assert (await client.get("/api/thermal")).status == 200  # cold: waits for the first fit
+        release, finished, calls = threading.Event(), threading.Event(), []
+        real = thermal.fit_sessions
+
+        def slow_fit(*args, **kwargs):
+            calls.append(1)
+            release.wait(10)
+            try:
+                return real(*args, **kwargs)
+            finally:
+                finished.set()
+
+        monkeypatch.setattr(thermal, "fit_sessions", slow_fit)
+        monkeypatch.setattr(web_module, "THERMAL_REFIT_S", 0.0)  # every request finds it stale
+        t0 = time.monotonic()
+        for _ in range(3):
+            assert (await client.get("/api/thermal")).status == 200
+        assert time.monotonic() - t0 < 2.0, "the stale fit is served without waiting"
+        assert len(calls) == 1, "one background refit, not one per request"
+        release.set()
+        await asyncio.to_thread(finished.wait, 10)
+
+
 async def test_backoff_on_unreachable_host(db, unused_tcp_port):
     cfg = Config(
         host=f"127.0.0.1:{unused_tcp_port}",

wallmonitor/thermal.py

@@ -276,6 +276,14 @@ def ambient_from_idle_handle(handle_c: float, model: IdleOffset = BUILTIN_IDLE_O
 # this split segments; each segment's opening ramp is a fit candidate.
 SEGMENT_SPLIT_GAP_S = 300.0
 MAX_SEGMENTS_PER_SESSION = 4
+# The forecast's tau, rise and current exponent come from the newest this
+# many sessions; the degradation watch uses every session in the lookback.
+# The cap predates any scoring. Lifting it for the forecast was backtested
+# (one install, 114 sessions): the median tau fell 11.5 -> 11.25 min and the
+# in-run trajectory forecast at cold starts moved 0.4-0.5 C optimistic,
+# the direction that trips the charger, for small gains elsewhere. So the
+# forecast keeps it until a deliberate window is scored on its own.
+FORECAST_SESSIONS = 40
 
 
 @dataclass
@@ -677,12 +685,22 @@ def fit_sessions(db: Database, now: float, lookback_days: float = 120.0) -> list
     back as the handle warmed, so the fitted plateau is a setpoint it held
     rather than the connector's own equilibrium. Such fits still serve the
     forecast — they describe what the handle actually did — but the
-    degradation watch compares only free-running ones."""
+    degradation watch compares only free-running ones.
+
+    Every session in the lookback is fitted, not just the newest few. The
+    degradation watch regresses on whatever this returns, so a count cap
+    makes its baseline whatever happened ~40 charges ago: on one install a
+    single odd week at the edge of a 40-session window read as +8 °C of
+    confirmed drift, where the full history since the verified anchor was
+    flat (-0.04 °C over 80 days). The time window, not a session count,
+    decides what the baseline is. Fits from the newest FORECAST_SESSIONS
+    carry forecast=True; only those teach the forecast (forecast_fits)."""
     sessions = [
         session
         for session in db.sessions_range(now - lookback_days * 86400, now)
         if session.get("end_ts") and (session.get("charging_s") or 0) >= MIN_SEGMENT_S
-    ][:40]
+    ]
+    forecast_ids = {session["id"] for session in sessions[:FORECAST_SESSIONS]}
     fits: list[dict] = []
     idle_model = load_idle_offset(db)
     for sess in sessions:
@@ -784,6 +802,7 @@ def fit_sessions(db: Database, now: float, lookback_days: float = 120.0) -> list
             fits.append(
                 {
                     "session_id": sess["id"],
+                    "forecast": sess["id"] in forecast_ids,
                     "start_ts": seg[0][0],
                     "tau_min": round(tau_s / 60.0, 2),
                     "rise_ref_c": round(rise, 2) if rise is not None else None,
@@ -810,6 +829,12 @@ def fit_sessions(db: Database, now: float, lookback_days: float = 120.0) -> list
     return fits
 
 
+def forecast_fits(fits: list[dict]) -> list[dict]:
+    """The fits the forecast's parameters come from: the newest
+    FORECAST_SESSIONS sessions' (see there for why not all of them)."""
+    return [fit for fit in fits if fit.get("forecast", True)]
+
+
 def _fit_current_exponent(fits: list[dict]) -> tuple[float, int, float | None]:
     """The exponent n in rise = rise_ref * (I/48)^n, from free-running fits:
     (n, fits used, standard error). The prior with no fits used when the
@@ -848,16 +873,21 @@ def fit_history(db: Database, now: float, lookback_days: float = 120.0,
     degradation watch, and the rise_ref_c median taken here, all share one
     current law. With the I^2 prior, off-reference fits carry a bias that the
     watch's current term then has to absorb; with a fitted exponent that
-    term has nothing left to explain."""
+    term has nothing left to explain.
+
+    The parameters, exponent included, come from forecast_fits only; the
+    re-normalization reaches every fit, so the watch's older fits share the
+    forecast's current law too."""
     if fits is None:
         fits = fit_sessions(db, now, lookback_days)
-    exponent, exp_fits, exp_se = _fit_current_exponent(fits)
+    recent = forecast_fits(fits)
+    exponent, exp_fits, exp_se = _fit_current_exponent(recent)
     for fit in fits:
         if fit.get("rise_c") is not None and fit.get("current_a"):
             fit["rise_ref_c"] = round(fit["rise_c"] * (REF_CURRENT_A / fit["current_a"]) ** exponent, 2)
-    taus = [fit["tau_min"] for fit in fits]
-    rises = [fit["rise_ref_c"] for fit in fits if fit["rise_ref_c"] is not None]
-    rmses = [fit["rmse_c"] for fit in fits]
+    taus = [fit["tau_min"] for fit in recent]
+    rises = [fit["rise_ref_c"] for fit in recent if fit["rise_ref_c"] is not None]
+    rmses = [fit["rmse_c"] for fit in recent]
     return ThermalParams(
         tau_min=median(taus) if taus else DEFAULT_TAU_MIN,
         rise_ref_c=median(rises) if rises else DEFAULT_RISE_REF_C,

wallmonitor/web.py

@@ -22,6 +22,9 @@
 
 STATIC_PKG = "wallmonitor.static"
 
+# How long /api/thermal serves a history fit before refitting in the background.
+THERMAL_REFIT_S = 6 * 3600
+
 
 def _float_q(request: web.Request, name: str, default: float) -> float:
     """Float query parameter, silently falling back on absent/garbage input
@@ -224,27 +227,60 @@ async def api_event_ingest(request: web.Request) -> web.Response:
 
     # Model parameters change only as new sessions land, so the (SQLite-heavy)
     # history fit is cached; the live prediction is computed on every call.
-    thermal_fit: dict = {"params": None, "fits": [], "ts": 0.0, "idle_offset": None}
+    # The fit covers the whole drift lookback and takes tens of seconds on a
+    # long history (27 s for 72 fits on a 4-core Intel N100), while the amp
+    # controller polls with a 10 s timeout. So a stale fit is served at once
+    # and refitted in the background; only a cold cache or ?refit waits.
+    thermal_fit: dict = {"params": None, "fits": [], "ts": 0.0, "idle_offset": None, "task": None}
+
+    async def refit_thermal() -> None:
+        started = time.time()
+        idle_offset = await asyncio.to_thread(db.get_setting, thermal.IDLE_OFFSET_SETTING)
+        fits = await asyncio.to_thread(thermal.fit_sessions, db, started)
+        params = await asyncio.to_thread(thermal.fit_history, db, started, fits=fits)
+        if started >= thermal_fit["ts"]:  # an older refit finishing late must not win
+            thermal_fit.update(fits=fits, params=params, ts=started, idle_offset=idle_offset)
+
+    def _log_refit_failure(task: asyncio.Task) -> None:
+        if not task.cancelled() and task.exception() is not None:
+            log.error("thermal refit failed", exc_info=task.exception())
+
+    def refit_in_background() -> asyncio.Task:
+        """The running refit, or a new one: never two at once."""
+        task = thermal_fit["task"]
+        if task is None or task.done():
+            task = asyncio.create_task(refit_thermal(), name="thermal-refit")
+            task.add_done_callback(_log_refit_failure)
+            thermal_fit["task"] = task
+        return task
+
+    async def warm_thermal(_app: web.Application) -> None:
+        refit_in_background()
+
+    async def stop_thermal(_app: web.Application) -> None:
+        task = thermal_fit["task"]
+        if task is not None and not task.done():
+            task.cancel()
+
+    app.on_startup.append(warm_thermal)
+    app.on_cleanup.append(stop_thermal)
 
     async def api_thermal(request: web.Request) -> web.Response:
         """The full thermal picture: fitted model, live forecast, per-segment
         fits, drift verdict, baseline anchor. Also what the BLE amp
-        controller polls every 30 s. ?refit busts the 6 h fit cache."""
+        controller polls every 30 s. ?refit waits for a fresh fit."""
         now = time.time()
         # A newly adopted idle-offset model reinterprets every proxy-tier
         # fit, so the cache is keyed on the stored model too.
         idle_offset = await asyncio.to_thread(db.get_setting, thermal.IDLE_OFFSET_SETTING)
-        if (
-            thermal_fit["params"] is None
-            or now - thermal_fit["ts"] > 6 * 3600
-            or thermal_fit["idle_offset"] != idle_offset
-            or "refit" in request.query
-        ):
-            fits = await asyncio.to_thread(thermal.fit_sessions, db, now)
-            thermal_fit["fits"] = fits
-            thermal_fit["params"] = thermal.fit_history(db, now, fits=fits)
-            thermal_fit["ts"] = now
-            thermal_fit["idle_offset"] = idle_offset
+        if "refit" in request.query:
+            await refit_thermal()
+        elif thermal_fit["params"] is None:
+            # Cold (just started): nothing to serve yet. Shielded so a caller
+            # that gives up doesn't cancel the refit everyone else is waiting on.
+            await asyncio.shield(refit_in_background())
+        elif now - thermal_fit["ts"] > THERMAL_REFIT_S or thermal_fit["idle_offset"] != idle_offset:
+            refit_in_background()
         result = await asyncio.to_thread(thermal.predict, db, now, thermal_fit["params"])
         anchor = await asyncio.to_thread(thermal.baseline_anchor, db)
         return web.json_response(