Weir rounds through the public Ephemeris API

.github/workflows/paracast-round.yml

@@ -14,8 +14,11 @@
 # date"), so the feed updates twice daily while history stays one round/day.
 #
 # SECRETS (Settings -> Secrets and variables -> Actions):
-#   PARACAST_URL            required — the deployed chute's base URL
-#   PARACAST_API_KEY        optional — bearer token if the gateway needs one
+#   PARACAST_URL            required — the public Ephemeris API
+#                           (https://ephemeris.cascade.industries/api/v1), or a raw
+#                           paracast backend's base URL
+#   PARACAST_API_KEY        a pc_live_ Ephemeris key (rounds are billed in credits),
+#                           or the raw backend's bearer token
 #   HIPPIUS_S3_ACCESS_KEY   required — pulls the scraped parquet mirror
 #   HIPPIUS_S3_SECRET_KEY   required
 #   HIPPIUS_S3_ENDPOINT     optional (default https://s3.hippius.com)
@@ -159,7 +162,24 @@ jobs:
           fi
           # ${arr[@]+...} so an empty auth array is not an unbound-variable
           # error under `set -u`.
-          probe() { curl -sf -m 30 ${auth[@]+"${auth[@]}"} "${PARACAST_URL}/health" 2>/dev/null; }
+          # The public Ephemeris API reports health at /api/status (no key), in
+          # its own shape; map it onto the raw /health fields used below.
+          case "${PARACAST_URL%/}" in
+            */api/v1)
+              status_url="${PARACAST_URL%/}"; status_url="${status_url%/api/v1}/api/status"
+              probe() {
+                curl -sf -m 30 "${status_url}" 2>/dev/null | python -c '
+          import json, sys
+          d = json.load(sys.stdin); m = d.get("models", [])
+          print(json.dumps({
+              "status": "ok" if d.get("state") == "operational" else d.get("state", "?"),
+              "healthy": [x["name"] for x in m if x.get("state") == "ready"],
+              "unhealthy": [x["name"] for x in m if x.get("enabled") and x.get("state") != "ready"],
+          }))'
+              } ;;
+            *)
+              probe() { curl -sf -m 30 ${auth[@]+"${auth[@]}"} "${PARACAST_URL}/health" 2>/dev/null; } ;;
+          esac
           field() { printf '%s' "$1" | python -c "import json,sys;d=json.load(sys.stdin);print($2)" 2>/dev/null; }
 
           up_wait="${PARACAST_UP_TIMEOUT:-1800}"   # 30 min to answer at all

src/paracast_client.py

@@ -22,6 +22,15 @@
   every benchmark round. The leaderboard and the serving weights improve from
   the same evaluation without either trusting the other's arithmetic.
 
+**Two front doors, one contract.** ``base_url`` is either the raw paracast
+backend (``https://host:port``: ``/forecast``, ``/models``, ``/health``,
+``/feedback``) or the public Ephemeris API (``https://…/api/v1``), which fronts
+the same backend for customers. The public API takes series as
+``{"values", "freq"}`` objects, answers ``/models`` with a bare list, serves
+health at ``/api/status``, bills every forecast in credits and accepts an
+``Idempotency-Key`` so a retried request is never charged twice. The client
+translates; scoring is unchanged either way.
+
 The client is stdlib-only (urllib) on top of the repo's numpy core: no httpx,
 no pydantic — the wire contract is mirrored here as plain dicts and pinned by
 tests/test_paracast_client.py against the shapes in paracast_common.
@@ -32,7 +41,9 @@
 import json
 import time
 import urllib.error
+import urllib.parse
 import urllib.request
+import uuid
 from dataclasses import dataclass, field
 
 import numpy as np
@@ -66,6 +77,10 @@
 # decile band being re-interpolated.
 REQUEST_QUANTILES: tuple[float, ...] = _CRPS_QUANTILES
 
+# Public Ephemeris API request limits (paracast-web lib/forecast/validation.ts).
+PUBLIC_MAX_CONTEXT = 16_384
+PUBLIC_MAX_OUTPUT_VALUES = 120_000  # series x horizon x quantiles per request
+
 # ISO-8601 sampling interval (challenge meta["freq"], the FREQ_SEASONALITY
 # keys) → the pandas-style alias paracast's feature extractor understands.
 # Unknown intervals send freq=None: paracast treats the hint as optional, and a
@@ -123,6 +138,8 @@ def __init__(
         transport=None,
     ):
         self.base_url = base_url.rstrip("/")
+        # The public Ephemeris API lives under /api/v1; anything else is the raw backend.
+        self.public = urllib.parse.urlparse(self.base_url).path.endswith("/api/v1")
         self.api_key = api_key
         self.timeout = timeout
         self.retries = retries
@@ -139,10 +156,18 @@ def _urllib_transport(self, method, url, body, headers):
         except urllib.error.HTTPError as e:  # non-2xx still carries a JSON body
             return e.code, e.read()
 
-    def _call(self, method: str, path: str, payload: dict | None = None) -> dict:
-        url = self.base_url + path
+    def _call(
+        self,
+        method: str,
+        path: str,
+        payload: dict | None = None,
+        *,
+        url: str | None = None,
+        headers: dict | None = None,
+    ) -> dict:
+        url = url or self.base_url + path
         body = None if payload is None else json.dumps(payload).encode()
-        headers = {"Content-Type": "application/json"}
+        headers = {"Content-Type": "application/json", **(headers or {})}
         if self.api_key:
             headers["Authorization"] = f"Bearer {self.api_key}"
         last: Exception | None = None
@@ -188,18 +213,42 @@ def forecast(
         }
         if model is not None:
             payload["model"] = model
-        if freq is not None:
-            payload["freq"] = freq
-        return self._call("POST", "/forecast", payload)
+        if not self.public:
+            if freq is not None:
+                payload["freq"] = freq
+            return self._call("POST", "/forecast", payload)
+        payload["series"] = [
+            {"values": s, **({"freq": freq} if freq is not None else {})} for s in series
+        ]
+        # The public API reads only the last context_len points (default 256); send
+        # the full history so rounds score exactly what the raw backend saw.
+        payload["context_len"] = min(max(len(s) for s in series), PUBLIC_MAX_CONTEXT)
+        # One key per logical request, reused by _call's retries: a retry after a
+        # timeout replays the stored result instead of paying for it twice.
+        key = f"weir-{uuid.uuid4().hex}"
+        return self._call("POST", "/forecast", payload, headers={"Idempotency-Key": key})
 
     def feedback(self, request_id: str, actuals: list[list[float]]) -> dict:
         return self._call("POST", "/feedback", {"request_id": request_id, "actuals": actuals})
 
     def models(self) -> dict:
-        return self._call("GET", "/models")
+        doc = self._call("GET", "/models")
+        # The public API answers with the provider's model list, unwrapped.
+        return {"models": doc} if isinstance(doc, list) else doc
 
     def health(self) -> dict:
-        return self._call("GET", "/health")
+        if not self.public:
+            return self._call("GET", "/health")
+        # Public status, mapped onto the raw /health shape the round expects.
+        root = self.base_url[: -len("/api/v1")]
+        doc = self._call("GET", "/api/status", url=root + "/api/status")
+        models = doc.get("models", [])
+        return {
+            "status": "ok" if doc.get("state") == "operational" else doc.get("state", "unknown"),
+            "healthy": [m["name"] for m in models if m.get("state") == "ready"],
+            "unhealthy": [m["name"] for m in models
+                          if m.get("enabled") and m.get("state") != "ready"],
+        }
 
     def panel(self) -> list[str]:
         """Names of the enabled+healthy serving panel, in manifest order."""
@@ -353,8 +402,12 @@ def prefetch_forecasts(
         table: dict[int, ProbForecast] = {}
         try:
             for (horizon, freq), indices in groups.items():
-                for start in range(0, len(indices), chunk):
-                    batch = indices[start : start + chunk]
+                step = chunk
+                if getattr(client, "public", False):  # stay inside the public API's output budget
+                    per_series = horizon * len(REQUEST_QUANTILES)
+                    step = max(1, min(chunk, PUBLIC_MAX_OUTPUT_VALUES // per_series))
+                for start in range(0, len(indices), step):
+                    batch = indices[start : start + step]
                     series = [_clean_series(challenges[i].context) for i in batch]
                     resp = client.forecast(
                         series,

tests/test_paracast_client.py

@@ -239,3 +239,112 @@ def test_models_json_covers_every_publishable_id():
         assert mid in registered, f"models.json lacks display metadata for {mid!r}"
     assert registered[pc.ENSEMBLE_ID]["class"] == "ensemble"
     assert registered[pc.ROUTER_ID]["class"] == "ensemble"
+
+
+# --------------------------------------------------------------------------- #
+# Public Ephemeris API front door (base_url ending in /api/v1)
+# --------------------------------------------------------------------------- #
+
+
+class PublicFakeEndpoint(FakeEndpoint):
+    """Answers like the Ephemeris gateway: /api/v1 routes, series objects,
+    an unwrapped /models list, health at /api/status. Forecasts are delegated
+    to the raw fake after translating the request back."""
+
+    def __call__(self, method, url, body, headers):
+        path = url.split("://", 1)[-1].split("/", 1)[1]
+        if path == "api/status":
+            self.calls.append({"method": method, "path": "/" + path,
+                               "payload": None, "headers": headers})
+            return 200, json.dumps({"state": "operational", "models": [
+                {"name": "chronos2", "state": "ready", "enabled": True},
+                {"name": "tirex2", "state": "warming", "enabled": True},
+                {"name": "moirai2", "state": "unavailable", "enabled": False},
+            ]}).encode()
+        assert path.startswith("api/v1/"), path
+        rest = path[len("api/v1/"):]
+        raw_url = "http://x/" + rest
+        if rest == "models":
+            status, raw = super().__call__(method, raw_url, body, headers)
+            return status, json.dumps(json.loads(raw)["models"]).encode()
+        if rest == "forecast":
+            payload = json.loads(body)
+            assert all(isinstance(s, dict) for s in payload["series"])
+            raw_payload = {**payload, "series": [s["values"] for s in payload["series"]]}
+            status, raw = super().__call__(method, raw_url, json.dumps(raw_payload).encode(),
+                                           headers)
+            self.calls[-1]["public_payload"] = payload
+            return status, raw
+        return super().__call__(method, raw_url, body, headers)
+
+
+def _public(fake, **kw):
+    return pc.ParacastClient("https://eph.example/api/v1/", transport=fake, retries=0, **kw)
+
+
+def test_public_flavour_is_detected_from_the_path():
+    assert _public(PublicFakeEndpoint()).public
+    assert not _client(FakeEndpoint()).public
+    assert not pc.ParacastClient("https://host:20004").public
+
+
+def test_public_forecast_sends_series_objects_full_context_and_a_key(challenges):
+    fake = PublicFakeEndpoint()
+    result = pc.prefetch_forecasts(_public(fake), challenges,
+                                   [pc.ModelSpec.explicit("chronos2")], verbose=False)
+    assert len(result.forecasts["chronos2"]) == 3
+    sent = [c for c in fake.calls if "public_payload" in c]
+    assert len(sent) == 2
+    for call in sent:
+        p = call["public_payload"]
+        assert "freq" not in p  # per series, not per request
+        assert {s["freq"] for s in p["series"]} <= {"h", "D"}
+        assert p["context_len"] == max(len(s["values"]) for s in p["series"])
+        assert call["headers"]["Idempotency-Key"].startswith("weir-")
+    assert len({c["headers"]["Idempotency-Key"] for c in sent}) == 2
+
+
+def test_public_retry_reuses_the_idempotency_key():
+    keys = []
+
+    def transport(method, url, body, headers):
+        keys.append(headers.get("Idempotency-Key"))
+        if len(keys) < 2:
+            return 503, b"busy"
+        return 200, json.dumps({"forecasts": [{"quantiles": {"0.5": [1.0]}}],
+                                "meta": {"request_id": "r"}}).encode()
+
+    client = pc.ParacastClient("https://eph.example/api/v1", transport=transport,
+                               retries=2, retry_wait=0.0)
+    client.forecast([[1.0, 2.0]], horizon=1)
+    assert len(keys) == 2 and keys[0] == keys[1] is not None
+
+
+def test_public_models_list_and_status_map_to_the_raw_shapes():
+    fake = PublicFakeEndpoint()
+    client = _public(fake)
+    assert client.panel() == ["chronos2", "tirex2"]
+    health = client.health()
+    assert health == {"status": "ok", "healthy": ["chronos2"], "unhealthy": ["tirex2"]}
+    assert fake.calls[-1]["path"] == "/api/status"
+
+
+def test_public_batches_stay_inside_the_output_budget():
+    horizon = 400  # 64 x 400 x 19 quantiles would be 486,400 values
+    chs = [_challenge(float(i), 32, horizon, "P1D", "src") for i in range(40)]
+    fake = PublicFakeEndpoint()
+    pc.prefetch_forecasts(_public(fake), chs, [pc.ModelSpec.explicit("chronos2")],
+                          verbose=False)
+    sizes = [len(c["public_payload"]["series"]) for c in fake.calls if "public_payload" in c]
+    cap = pc.PUBLIC_MAX_OUTPUT_VALUES // (horizon * len(pc.REQUEST_QUANTILES))
+    assert sum(sizes) == 40 and max(sizes) == cap
+
+
+def test_public_feedback_goes_to_the_gateway_route(challenges):
+    fake = PublicFakeEndpoint()
+    client = _public(fake)
+    result = pc.prefetch_forecasts(client, challenges,
+                                   [pc.ModelSpec.explicit("chronos2")], verbose=False)
+    out = pc.send_feedback(client, challenges, result.requests, verbose=False)
+    assert out == {"sent": 2, "failed": 0, "skipped": 0}
+    assert len(fake.feedbacks) == 2