diff --git a/.github/workflows/paracast-round.yml b/.github/workflows/paracast-round.yml index 5666f2d..610e051 100644 --- a/.github/workflows/paracast-round.yml +++ b/.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 diff --git a/src/paracast_client.py b/src/paracast_client.py index b27c27e..94061e4 100644 --- a/src/paracast_client.py +++ b/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, diff --git a/tests/test_paracast_client.py b/tests/test_paracast_client.py index 83585cb..dd20a54 100644 --- a/tests/test_paracast_client.py +++ b/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