Stones C6: add step_ttm_battery_surge specialist

Wraps msradam/Granite-TTM-r2-Battery-Surge — Apache-2.0 fine-tune of
ibm-granite/granite-timeseries-ttm-r2 trained on AMD MI300X via AMD
Developer Cloud. Test MAE 0.1091 m on held-out 2023-2024 windows
(-41% vs persistence, -25% vs zero-shot TTM r2).

app/live/ttm_battery_surge.py
fetch() pulls the trailing 1024 h of hourly verified water level +
harmonic tide predictions from NOAA station 8518750 (The Battery),
computes surge residual, runs the fine-tuned TTM, and returns the
96 h forecast plus peak-residual / peak-time scalars the reconciler
can paraphrase. Gated by RIPRAP_TTM_BATTERY_SURGE_ENABLE.

app/fsm.py
New @action step_ttm_battery_surge wired into build_app's actions
dict next to the existing TTM specialists. Plumbed through
step_reconcile reads, the snap dict, run() output, and
iter_steps' final-state dict so the SSE stream surfaces the new
state key end-to-end.

app/reconcile.py
build_documents() emits a `ttm_battery` doc in the Lodestone block
(after floodnet_forecast, before the policy section). Doc body
paraphrases the source/citation, context window, horizon, recent
residual, peak forecast, and peak-time-UTC. Gated on
interesting=True so calm-day forecasts stay silent.
trim_docs_to_plan PREFIXES_BY_SPECIALIST gains a ttm_battery_surge
-> ttm_battery mapping.

web/static/agent.js
STEP_LABELS / SOURCE_LABELS / SOURCE_URLS / SOURCE_VINTAGES extended
for the new specialist; the Source URL points at the published HF
artifact.

Doc-order snapshot stays clean — `ttm_battery` lands last in Lodestone,
no doc_id collisions, no other Stone groups perturbed.

Both TTM specialists (zero-shot ttm_forecast, fine-tuned
ttm_battery_surge) coexist intentionally — different cadences, different
horizons, different gauges.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

app/fsm.py

@@ -499,6 +499,44 @@ def step_ttm_forecast(state: State) -> State:
         rec["elapsed_s"] = round(time.time() - rec["started_at"], 2)
 
 
+@action(reads=["lat", "lon"], writes=["ttm_battery_surge", "trace"])
+def step_ttm_battery_surge(state: State) -> State:
+    """Granite TTM r2 fine-tune — 96 h hourly Battery surge nowcast.
+
+    Same TTM r2 backbone family as step_ttm_forecast but a different
+    artefact: msradam/Granite-TTM-r2-Battery-Surge, trained on AMD
+    MI300X. Hourly cadence vs the zero-shot's 6-min, 4-day vs 9.6 h
+    horizon. Both can fire on the same query — the reconciler frames
+    each as a distinct forecast in the briefing."""
+    rec, trace = _step(state, "ttm_battery_surge")
+    try:
+        if state.get("lat") is None:
+            rec["ok"] = False; rec["err"] = "no coords"
+            return state.update(ttm_battery_surge=None, trace=trace)
+        # Battery gauge is a single point; the forecast applies citywide
+        # to NYC harbor entrance, so we don't gate by NYC bbox.
+        from app.live import ttm_battery_surge
+        s = ttm_battery_surge.fetch()
+        rec["ok"] = bool(s.get("available"))
+        if not rec["ok"]:
+            rec["err"] = s.get("reason", "unavailable")
+            return state.update(ttm_battery_surge=None, trace=trace)
+        rec["result"] = {
+            "context_h": s.get("context_hours"),
+            "horizon_h": s.get("horizon_hours"),
+            "forecast_peak_m": s.get("forecast_peak_m"),
+            "forecast_peak_hours_ahead": s.get("forecast_peak_hours_ahead"),
+            "interesting": s.get("interesting"),
+        }
+        return state.update(ttm_battery_surge=s, trace=trace)
+    except Exception as e:
+        rec["ok"] = False; rec["err"] = str(e)
+        log.exception("ttm_battery_surge failed")
+        return state.update(ttm_battery_surge=None, trace=trace)
+    finally:
+        rec["elapsed_s"] = round(time.time() - rec["started_at"], 2)
+
+
 @action(reads=["lat", "lon"], writes=["floodnet_forecast", "trace"])
 def step_floodnet_forecast(state: State) -> State:
     """TTM r2 forecast of flood-event recurrence at the nearest FloodNet
@@ -893,7 +931,8 @@ def _label_counts(gliner_out: dict[str, dict]) -> dict[str, int]:
                "ida_hwm", "prithvi_water", "prithvi_live", "terramind",
                "terramind_lulc", "terramind_buildings",
                "noaa_tides", "nws_alerts", "nws_obs", "ttm_forecast",
-               "ttm_311_forecast", "floodnet_forecast", "mta_entrances",
+               "ttm_311_forecast", "floodnet_forecast", "ttm_battery_surge",
+               "mta_entrances",
                "nycha_developments", "doe_schools", "doh_hospitals",
                "rag", "gliner"],
         writes=["paragraph", "audit", "mellea", "trace"])
@@ -917,6 +956,7 @@ def step_reconcile(state: State) -> State:
             "ttm_forecast": state.get("ttm_forecast"),
             "ttm_311_forecast": state.get("ttm_311_forecast"),
             "floodnet_forecast": state.get("floodnet_forecast"),
+            "ttm_battery_surge": state.get("ttm_battery_surge"),
             "rag": state.get("rag"),
             "gliner": state.get("gliner"),
             "prithvi_live": state.get("prithvi_live"),
@@ -1027,6 +1067,7 @@ def build_app(query: str):
         "ttm_forecast": step_ttm_forecast,
         "ttm_311_forecast": step_ttm_311_forecast,
         "floodnet_forecast": step_floodnet_forecast,
+        "ttm_battery_surge": step_ttm_battery_surge,
         "microtopo": step_microtopo,
         "ida_hwm": step_ida_hwm,
         "mta_entrances": step_mta_entrances,
@@ -1091,6 +1132,7 @@ def run(query: str) -> dict[str, Any]:
         "ttm_forecast": final_state.get("ttm_forecast"),
         "ttm_311_forecast": final_state.get("ttm_311_forecast"),
         "floodnet_forecast": final_state.get("floodnet_forecast"),
+        "ttm_battery_surge": final_state.get("ttm_battery_surge"),
         "mta_entrances": final_state.get("mta_entrances"),
         "nycha_developments": final_state.get("nycha_developments"),
         "doe_schools": final_state.get("doe_schools"),
@@ -1214,6 +1256,7 @@ def iter_steps(query: str):
         "ttm_forecast": state.get("ttm_forecast"),
         "ttm_311_forecast": state.get("ttm_311_forecast"),
         "floodnet_forecast": state.get("floodnet_forecast"),
+        "ttm_battery_surge": state.get("ttm_battery_surge"),
         "mta_entrances": state.get("mta_entrances"),
         "nycha_developments": state.get("nycha_developments"),
         "doe_schools": state.get("doe_schools"),

app/live/ttm_battery_surge.py

@@ -0,0 +1,279 @@
+"""Granite TTM r2 — Battery 96 h surge nowcast (NYC fine-tune).
+
+Wraps the Apache-2.0 [`msradam/Granite-TTM-r2-Battery-Surge`](https://huggingface.co/msradam/Granite-TTM-r2-Battery-Surge)
+fine-tune. Fetches the past 1024 hours (~43 days) of hourly verified
+water level + harmonic tide predictions at NOAA station 8518750 (The
+Battery), computes surge residual (observed − predicted), and forecasts
+the next 96 hours.
+
+Distinct from `app.live.ttm_forecast` — that's the *zero-shot* TTM r2
+on 6-min cadence (~9.6 h horizon) at the closest of three NYC gauges.
+This module is the *fine-tuned* model on hourly cadence (~4-day horizon)
+at a single gauge (Battery only — see MODEL_CARD honest-limitations).
+
+Both nowcasts coexist in the FSM. The zero-shot is shorter-horizon and
+covers every coastal NYC query; the fine-tuned is longer-horizon and
+specialised to the Battery's storm-surge regime, which is the dominant
+driver of NYC inundation. The reconciler frames each as a separate
+forecast in the briefing.
+
+Gated by RIPRAP_TTM_BATTERY_SURGE_ENABLE — deployments without the
+heavy ML deps (granite-tsfm / transformers) silently no-op via the
+same skipped-result shape every other heavy specialist emits.
+"""
+from __future__ import annotations
+
+import logging
+import os
+import threading
+import time
+from datetime import datetime, timedelta
+from typing import Any
+
+log = logging.getLogger("riprap.ttm_battery_surge")
+
+ENABLE = os.environ.get(
+    "RIPRAP_TTM_BATTERY_SURGE_ENABLE", "1"
+).lower() in ("1", "true", "yes")
+DEVICE = os.environ.get("RIPRAP_TTM_BATTERY_SURGE_DEVICE", "cpu")
+REPO = "msradam/Granite-TTM-r2-Battery-Surge"
+
+DOC_ID = "ttm_battery"
+CITATION = (
+    "msradam/Granite-TTM-r2-Battery-Surge (Apache-2.0, fine-tune of "
+    "ibm-granite/granite-timeseries-ttm-r2). Trained on AMD Instinct "
+    "MI300X via AMD Developer Cloud. Test MAE 0.1091 m on held-out "
+    "2023-2024 windows (vs 0.1467 zero-shot, 0.1861 persistence)."
+)
+
+# NOAA Battery (NY) — the canonical NYC storm-surge gauge.
+STATION_ID = "8518750"
+STATION_NAME = "The Battery, NY"
+NOAA_API = "https://api.tidesandcurrents.noaa.gov/api/prod/datagetter"
+
+# TTM r2 1024-96-r2 backbone: 1024 hours of context, 96 hours of horizon.
+CONTEXT_LENGTH = 1024
+PREDICTION_LENGTH = 96
+
+# Doc emission gate: only cite the forecast if the predicted peak surge
+# is meaningful (positive ≥0.3 m or negative ≤-0.3 m). On a calm day the
+# model still runs but the reconciler sees no doc.
+MIN_INTERESTING_RESIDUAL_M = float(
+    os.environ.get("RIPRAP_TTM_BATTERY_MIN_INTERESTING_M", "0.3"))
+
+_MODEL = None
+_INIT_LOCK = threading.Lock()
+
+
+def _has_required_deps() -> tuple[bool, str | None]:
+    missing: list[str] = []
+    for name in ("tsfm_public", "huggingface_hub", "torch", "requests",
+                 "pandas"):
+        try:
+            __import__(name)
+        except ImportError:
+            missing.append(name)
+    if missing:
+        return False, ", ".join(missing)
+    return True, None
+
+
+_DEPS_OK, _DEPS_MISSING = _has_required_deps()
+
+
+def _ensure_model():
+    """Load the fine-tuned TTM r2 once and cache. Failure is sticky —
+    a downloaded-then-broken model leaves _MODEL=None so subsequent
+    fetches re-attempt rather than silently serving a half-built one."""
+    global _MODEL
+    if _MODEL is not None:
+        return _MODEL
+    with _INIT_LOCK:
+        if _MODEL is not None:
+            return _MODEL
+        from huggingface_hub import snapshot_download
+        from tsfm_public import TinyTimeMixerForPrediction
+        log.info("ttm_battery_surge: downloading %s", REPO)
+        local_dir = snapshot_download(REPO)
+        log.info("ttm_battery_surge: loading model from %s", local_dir)
+        model = TinyTimeMixerForPrediction.from_pretrained(local_dir).eval()
+        if DEVICE == "cuda":
+            try:
+                import torch
+                if torch.cuda.is_available():
+                    model = model.cuda()
+            except Exception:
+                log.exception("ttm_battery_surge: cuda move failed; "
+                              "staying on CPU")
+        _MODEL = model
+        return _MODEL
+
+
+def _fetch_chunk(start: datetime, end: datetime, product: str):
+    """Pull one ≤30-day chunk from the NOAA CO-OPS datagetter.
+
+    Two products: `water_level` (verified, 6-min — we ask for hourly
+    via interval=h) and `predictions` (hourly harmonic tide). Both come
+    back in metres if `units=metric`.
+    """
+    import pandas as pd
+    import requests
+    params = {
+        "station": STATION_ID,
+        "begin_date": start.strftime("%Y%m%d"),
+        "end_date": end.strftime("%Y%m%d"),
+        "product": product,
+        "datum": "MLLW",
+        "units": "metric",
+        "time_zone": "gmt",
+        "format": "json",
+        "application": "riprap-nyc",
+        "interval": "h",
+    }
+    resp = requests.get(NOAA_API, params=params, timeout=30)
+    resp.raise_for_status()
+    data = resp.json()
+    key = "data" if "data" in data else "predictions"
+    if key not in data:
+        return pd.DataFrame()
+    df = pd.DataFrame(data[key])
+    if df.empty:
+        return df
+    df["timestamp"] = pd.to_datetime(df["t"])
+    df["value"] = pd.to_numeric(df["v"], errors="coerce")
+    return df[["timestamp", "value"]].dropna()
+
+
+def _fetch_battery_history(hours: int) -> Any:
+    """Pull the last `hours` hours of (water_level, predicted) at the
+    Battery and return a DataFrame with columns
+    `timestamp / water_level_m / predicted_m / surge_residual_m`."""
+    import pandas as pd
+
+    end_d = datetime.utcnow().replace(minute=0, second=0, microsecond=0)
+    n_days = max(1, hours // 24 + 3)  # padding in case of NOAA gaps
+
+    chunks_wl, chunks_pr = [], []
+    cur = end_d - timedelta(days=n_days)
+    while cur < end_d:
+        nxt = min(cur + timedelta(days=30), end_d)
+        try:
+            chunks_wl.append(_fetch_chunk(cur, nxt, "water_level"))
+            chunks_pr.append(_fetch_chunk(cur, nxt, "predictions"))
+        except Exception as e:
+            log.warning("ttm_battery_surge: NOAA chunk %s..%s failed: %s",
+                        cur.date(), nxt.date(), e)
+        cur = nxt
+
+    wl = pd.concat(chunks_wl, ignore_index=True) if chunks_wl else pd.DataFrame()
+    pr = pd.concat(chunks_pr, ignore_index=True) if chunks_pr else pd.DataFrame()
+    if wl.empty or pr.empty:
+        return pd.DataFrame()
+    wl = wl.rename(columns={"value": "water_level_m"})
+    pr = pr.rename(columns={"value": "predicted_m"})
+    df = wl.merge(pr, on="timestamp", how="inner").sort_values("timestamp")
+    df["surge_residual_m"] = df["water_level_m"] - df["predicted_m"]
+    df = df.dropna(subset=["surge_residual_m"])
+    if len(df) > hours:
+        df = df.iloc[-hours:].reset_index(drop=True)
+    return df
+
+
+def _summarize(history_df, forecast_arr) -> dict[str, Any]:
+    """Build the public dict the FSM specialist hands to the reconciler.
+
+    Includes both raw arrays (for downstream charts in the trace UI)
+    and human-readable scalars (peak / peak time / interesting flag)
+    that the reconciler can paraphrase without overshooting evidence.
+    """
+    import numpy as np
+    history_arr = history_df["surge_residual_m"].to_numpy()
+    history_recent = float(history_arr[-1]) if len(history_arr) else None
+    history_peak_abs = float(np.max(np.abs(history_arr))) if len(history_arr) else None
+
+    fc = np.asarray(forecast_arr, dtype="float64").reshape(-1)
+    if fc.size == 0:
+        return {"available": False, "reason": "empty forecast"}
+    peak_idx = int(np.argmax(np.abs(fc)))
+    peak = float(fc[peak_idx])
+    peak_h_ahead = peak_idx + 1  # hourly cadence; index 0 = +1 h
+
+    last_ts = (history_df["timestamp"].iloc[-1]
+               if len(history_df) else datetime.utcnow())
+    peak_time = last_ts + timedelta(hours=peak_h_ahead)
+
+    interesting = bool(abs(peak) >= MIN_INTERESTING_RESIDUAL_M)
+
+    return {
+        "available": True,
+        "interesting": interesting,
+        "model": REPO,
+        "station_id": STATION_ID,
+        "station_name": STATION_NAME,
+        "context_hours": int(len(history_arr)),
+        "horizon_hours": int(fc.size),
+        "history_recent_m": (round(history_recent, 3)
+                              if history_recent is not None else None),
+        "history_peak_abs_m": (round(history_peak_abs, 3)
+                                if history_peak_abs is not None else None),
+        "forecast_peak_m": round(peak, 3),
+        "forecast_peak_hours_ahead": peak_h_ahead,
+        "forecast_peak_time_utc": peak_time.isoformat(timespec="minutes"),
+        "forecast_array_m": [round(float(v), 4) for v in fc.tolist()],
+    }
+
+
+def fetch(timeout_s: float = 60.0) -> dict[str, Any]:
+    """Run the specialist. Always returns a dict with at minimum
+    `{available: bool, reason | ...}`. Caller should treat
+    `available=False` as silence-over-confabulation."""
+    if not ENABLE:
+        return {"available": False,
+                "reason": "RIPRAP_TTM_BATTERY_SURGE_ENABLE=0"}
+    if not _DEPS_OK:
+        return {"available": False,
+                "reason": f"deps unavailable on this deployment: "
+                          f"{_DEPS_MISSING}"}
+    t0 = time.time()
+    try:
+        df = _fetch_battery_history(CONTEXT_LENGTH)
+        if len(df) < CONTEXT_LENGTH:
+            return {"available": False,
+                    "reason": f"insufficient NOAA history: "
+                              f"got {len(df)} hours, need {CONTEXT_LENGTH}"}
+        if time.time() - t0 > timeout_s:
+            return {"available": False,
+                    "reason": "NOAA fetch exceeded budget"}
+
+        import torch
+        model = _ensure_model()
+        # [B=1, T=1024, C=1] tensor of metres surge residual.
+        residuals = df["surge_residual_m"].to_numpy().astype("float32")
+        past = torch.from_numpy(residuals).unsqueeze(0).unsqueeze(-1)
+        if DEVICE == "cuda":
+            try:
+                if torch.cuda.is_available():
+                    past = past.cuda()
+            except Exception:
+                log.exception("ttm_battery_surge: cuda move failed")
+        with torch.no_grad():
+            out = model(past_values=past)
+        forecast = out.prediction_outputs.squeeze(-1).squeeze(0).cpu().numpy()
+        result = _summarize(df, forecast)
+        result["elapsed_s"] = round(time.time() - t0, 2)
+        return result
+    except Exception as e:
+        log.exception("ttm_battery_surge fetch failed")
+        return {"available": False,
+                "reason": f"{type(e).__name__}: {e}",
+                "elapsed_s": round(time.time() - t0, 2)}
+
+
+def warm():
+    """Optional pre-load — amortizes the first-query model build cost."""
+    if not ENABLE or not _DEPS_OK:
+        return
+    try:
+        _ensure_model()
+    except Exception:
+        log.exception("ttm_battery_surge: warm() failed")

app/reconcile.py

@@ -270,6 +270,7 @@ def trim_docs_to_plan(doc_msgs: list[dict],
         "ttm_forecast":       ("ttm_forecast",),
         "ttm_311_forecast":   ("ttm_311_forecast",),
         "floodnet_forecast":  ("floodnet_forecast",),
+        "ttm_battery_surge":  ("ttm_battery",),
         "terramind":          ("terramind", "syn_"),
         "terramind_lulc":     ("tm_lulc",),
         "terramind_buildings": ("tm_buildings",),
@@ -1000,6 +1001,39 @@ def build_documents(state: dict[str, Any]) -> list[dict]:
         ]
         docs.append(_doc_message(doc_id, body))
 
+    # Granite TTM r2 — Battery surge fine-tune (msradam/Granite-TTM-r2-
+    # Battery-Surge, Apache-2.0, fine-tuned on AMD MI300X). Hourly
+    # cadence, 96 h horizon — distinct from the existing zero-shot
+    # ttm_forecast above, which runs at 6-min cadence over a 9.6 h
+    # horizon. Both can fire on the same query.
+    tbs = state.get("ttm_battery_surge")
+    if (not out_of_nyc and tbs and tbs.get("available")
+            and tbs.get("interesting")):
+        body = [
+            "Source: msradam/Granite-TTM-r2-Battery-Surge (Apache-2.0). "
+            "Fine-tune of ibm-granite/granite-timeseries-ttm-r2 trained "
+            "on AMD Instinct MI300X via AMD Developer Cloud. Test MAE "
+            "0.1091 m, -41% vs persistence and -25% vs zero-shot TTM r2.",
+            f"Gauge: {tbs['station_name']} (NOAA {tbs['station_id']}).",
+            f"Context window: {tbs['context_hours']} hours "
+            f"(~{tbs['context_hours']/24:.1f} days) of hourly surge "
+            "residual (verified water level minus harmonic tide).",
+            f"Forecast horizon: {tbs['horizon_hours']} hours "
+            f"(~{tbs['horizon_hours']/24:.1f} days ahead).",
+            f"Recent residual: {tbs['history_recent_m']} m.",
+            f"Recent peak |residual| in context: "
+            f"{tbs['history_peak_abs_m']} m.",
+            f"Forecast peak surge residual: {tbs['forecast_peak_m']} m, "
+            f"expected {tbs['forecast_peak_hours_ahead']} hours from "
+            f"now (at {tbs['forecast_peak_time_utc']} UTC).",
+            "INTERPRETATION: positive residual is the meteorological "
+            "component (storm surge, atmospheric pressure, wind setup) "
+            "on top of astronomical tide. The Battery is the dominant "
+            "NYC harbor-entrance gauge — its surge characterises Sandy "
+            "and Ida conditions citywide.",
+        ]
+        docs.append(_doc_message("ttm_battery", body))
+
     # ---- Policy context (RAG + GLiNER, ancillary to the four Stones) ---
     # Retrieved policy paragraphs and GLiNER typed-entity extractions.
     # These don't belong to a specific Stone — they ground the

web/static/agent.js

@@ -16,6 +16,7 @@ const STEP_LABELS = {
   ttm_forecast:           ["Granite TTM r2 — surge nowcast",   "9.6h forecast at the closest of Battery / Kings Pt / Sandy Hook"],
   ttm_311_forecast:       ["Granite TTM r2 — 311 forecast",    "4-week per-address flood-complaint forecast (52w history)"],
   floodnet_forecast:      ["Granite TTM r2 — FloodNet forecast", "flood-event recurrence forecast at nearest FloodNet sensor"],
+  ttm_battery_surge:      ["Granite TTM r2 — Battery surge (NYC fine-tune)", "96 h hourly surge nowcast at NOAA Battery (msradam/Granite-TTM-r2-Battery-Surge)"],
   mta_entrance_exposure:  ["MTA subway entrances",              "subway-entrance exposure (point-in-polygon Sandy + DEP)"],
   nycha_development_exposure: ["NYCHA developments",            "NYCHA campus footprint × Sandy + DEP overlap %"],
   doe_school_exposure:    ["NYC DOE schools",                   "school-point exposure (Sandy + DEP)"],
@@ -82,6 +83,7 @@ const SOURCE_LABELS = {
   ttm_forecast: "Granite TimeSeries TTM r2 — surge residual nowcast",
   ttm_311_forecast: "Granite TimeSeries TTM r2 — per-address 311 weekly forecast",
   floodnet_forecast: "Granite TimeSeries TTM r2 — FloodNet sensor recurrence forecast",
+  ttm_battery: "Granite TTM r2 NYC fine-tune — 96 h Battery surge nowcast (msradam/Granite-TTM-r2-Battery-Surge)",
   dob_permits: "NYC DOB Permit Issuance (Socrata ipu4-2q9a)",
   live_target: "Riprap planner — live target",
   rag_comptroller: 'NYC Comptroller — "Is NYC Ready for Rain?" (2024)',
@@ -133,6 +135,7 @@ const SOURCE_URLS = {
   ttm_forecast:           "https://huggingface.co/ibm-granite/granite-timeseries-ttm-r2",
   ttm_311_forecast:       "https://huggingface.co/ibm-granite/granite-timeseries-ttm-r2",
   floodnet_forecast:      "https://huggingface.co/ibm-granite/granite-timeseries-ttm-r2",
+  ttm_battery:            "https://huggingface.co/msradam/Granite-TTM-r2-Battery-Surge",
   dob_permits:            "https://data.cityofnewyork.us/Housing-Development/DOB-Permit-Issuance/ipu4-2q9a",
   rag_comptroller:        "https://comptroller.nyc.gov/reports/is-new-york-city-ready-for-rain/",
   rag_npcc4:              "https://nyaspubs.onlinelibrary.wiley.com/toc/17496632/2024/1539/1",
@@ -181,6 +184,7 @@ const SOURCE_VINTAGES = {
   ttm_forecast:           "live TTM forecast based on trailing 51 h at the closest NOAA gauge to this address (Battery / Kings Pt / Sandy Hook)",
   ttm_311_forecast:       "live TTM forecast based on trailing 52 weeks of NYC 311 flood complaints within 200 m of this address",
   floodnet_forecast:      "live TTM forecast based on the 512-day daily flood-event series at the nearest FloodNet sensor",
+  ttm_battery:            "live NYC fine-tuned TTM forecast based on the trailing 1024 hours (~43 days) of hourly surge residual at the Battery; 96 h horizon",
   dob_permits:            "live NYC DOB Permit Issuance, trailing 18-month window (per-permit issuance dates in payload)",
   rag_comptroller:        "NYC Comptroller report 'Is NYC Ready for Rain?' (2024)",
   rag_npcc4:              "NPCC4 — NYC Climate Assessment 4th edition, Annals NYAS vol. 1539 (2024)",