src/tsfm_bench/eval/external_api_predictor.py

from __future__ import annotations

import hashlib
import logging
import os
import time
from dataclasses import dataclass
from typing import Any, Iterator, List, Optional
from urllib.parse import urlparse

import httpx
import numpy as np
from gluonts.dataset import Dataset as GluonDataset
from gluonts.model import Forecast
from gluonts.model.forecast import QuantileForecast
from gluonts.model.predictor import RepresentablePredictor

from tsfm_bench.eval.predictors import _ForecastConfig

logger = logging.getLogger(__name__)

DEFAULT_EXTERNAL_TIMEOUT = 90.0
DEFAULT_MAX_CONTEXT_POINTS = 4096
DEFAULT_MAX_RESPONSE_BYTES = 5 * 1024 * 1024
RETRYABLE_STATUS_CODES = {408, 429, 500, 502, 503, 504}


@dataclass
class ExternalApiConfig:
    """Configuration for a user-hosted community forecasting endpoint."""

    endpoint_url: str
    model_id: str
    auth_token_env: str | None = None
    auth_header: str = "Authorization"
    timeout: float = DEFAULT_EXTERNAL_TIMEOUT
    max_retries: int = 2
    max_context_points: int = DEFAULT_MAX_CONTEXT_POINTS
    max_response_bytes: int = DEFAULT_MAX_RESPONSE_BYTES
    require_https: bool = True
    send_item_metadata: bool = False


class ExternalApiPredictor(RepresentablePredictor):
    """Forecast predictor backed by a user-operated HTTPS endpoint.

    This adapter intentionally treats the endpoint as a black box. The
    leaderboard process never imports user code or downloads user weights; it
    only sends the causal context window and validates the returned forecasts.
    """

    def __init__(
        self,
        config: ExternalApiConfig,
        prediction_length: int,
        quantile_levels: Optional[List[float]] = None,
        *,
        transport: httpx.BaseTransport | None = None,
    ):
        super().__init__(prediction_length=prediction_length)
        _validate_endpoint_url(config.endpoint_url, require_https=config.require_https)
        self.config = config
        self.quantile_levels = quantile_levels or []
        self.forecast_config = _ForecastConfig.from_quantiles(self.quantile_levels)
        self.leaderboard_name = config.model_id
        self._transport = transport

    def predict(self, dataset: GluonDataset, **kwargs) -> Iterator[Forecast]:
        headers = {"Content-Type": "application/json"}
        token = _read_auth_token(self.config.auth_token_env)
        if token:
            headers[self.config.auth_header] = f"Bearer {token}"

        with httpx.Client(
            timeout=self.config.timeout,
            trust_env=False,
            transport=self._transport,
        ) as client:
            dataset_freq = getattr(dataset, "freq", None)
            for entry in dataset:
                target = np.asarray(entry["target"], dtype=np.float64)
                univariate = target.ndim == 1
                working = target.reshape(1, -1) if univariate else target

                freq = _entry_frequency(entry, dataset_freq)
                start = _format_start(entry.get("start"))
                forecast_rows = []
                for variate_idx in range(working.shape[0]):
                    series = _finite_tail(
                        working[variate_idx],
                        max_points=self.config.max_context_points,
                    )
                    payload = _build_request_payload(
                        config=self.config,
                        series=series,
                        prediction_length=self.prediction_length,
                        quantile_levels=self.quantile_levels,
                        freq=freq,
                        start=start,
                        item_id=str(entry.get("item_id", variate_idx)),
                        variate_idx=variate_idx,
                    )
                    response_payload = _post_with_retries(
                        client,
                        self.config,
                        headers=headers,
                        payload=payload,
                    )
                    forecast_rows.append(
                        parse_external_forecast(
                            response_payload,
                            prediction_length=self.prediction_length,
                            quantile_levels=self.quantile_levels,
                        )
                    )

                stacked = np.stack(forecast_rows, axis=0)
                forecast_arrays = stacked[0] if univariate else stacked
                yield QuantileForecast(
                    forecast_arrays=forecast_arrays,
                    forecast_keys=self.forecast_config.forecast_keys,
                    start_date=entry["start"] + target.shape[-1],
                    item_id=entry["item_id"],
                )


def _validate_endpoint_url(endpoint_url: str, *, require_https: bool = True) -> None:
    parsed = urlparse(endpoint_url)
    if parsed.scheme not in {"http", "https"}:
        raise ValueError("External model endpoint must use http:// or https://")
    if not parsed.netloc:
        raise ValueError("External model endpoint URL is missing a host")
    host = parsed.hostname or ""
    local_host = host in {"localhost", "127.0.0.1", "::1"}
    if require_https and parsed.scheme != "https" and not local_host:
        raise ValueError("External model endpoint must use HTTPS")


def _read_auth_token(env_name: str | None) -> str | None:
    if not env_name:
        return None
    token = os.getenv(env_name)
    return token.strip() if token else None


def _entry_frequency(entry: dict[str, Any], dataset_freq: object | None) -> str:
    if dataset_freq:
        return str(dataset_freq)
    if entry.get("freq"):
        return str(entry["freq"])
    start = entry.get("start")
    return str(getattr(start, "freqstr", None) or "H")


def _format_start(start: Any) -> str:
    if hasattr(start, "to_timestamp"):
        return start.to_timestamp().isoformat()
    return str(start)


def _finite_tail(series: np.ndarray, *, max_points: int) -> np.ndarray:
    values = np.asarray(series, dtype=np.float64).reshape(-1)
    values = values[np.isfinite(values)]
    if values.size == 0:
        values = np.zeros(1, dtype=np.float64)
    if values.size > max_points:
        values = values[-max_points:]
    return values


def _opaque_item_id(item_id: str, variate_idx: int) -> str:
    digest = hashlib.sha256(f"{item_id}:{variate_idx}".encode("utf-8")).hexdigest()[:16]
    return f"series-{digest}"


def _build_request_payload(
    *,
    config: ExternalApiConfig,
    series: np.ndarray,
    prediction_length: int,
    quantile_levels: list[float],
    freq: str,
    start: str,
    item_id: str,
    variate_idx: int,
) -> dict[str, Any]:
    input_block: dict[str, Any] = {
        "series_id": _opaque_item_id(item_id, variate_idx),
        "target": [float(value) for value in series.tolist()],
    }
    if config.send_item_metadata:
        input_block.update({"item_id": item_id, "start": start, "freq": freq})

    return {
        "protocol_version": "tsfm-realworld-v1",
        "model": config.model_id,
        "inputs": [input_block],
        "parameters": {
            "prediction_length": int(prediction_length),
            "freq": freq,
            "quantiles": [float(level) for level in quantile_levels],
        },
    }


def _post_with_retries(
    client: httpx.Client,
    config: ExternalApiConfig,
    *,
    headers: dict[str, str],
    payload: dict[str, Any],
) -> dict[str, Any]:
    attempts = max(1, int(config.max_retries) + 1)
    last_error: Exception | None = None
    for attempt in range(1, attempts + 1):
        try:
            response = client.post(config.endpoint_url, headers=headers, json=payload)
            if response.status_code in RETRYABLE_STATUS_CODES and attempt < attempts:
                _sleep_before_retry(attempt)
                continue
            response.raise_for_status()
            if len(response.content) > config.max_response_bytes:
                raise ValueError(
                    f"External endpoint response is too large: "
                    f"{len(response.content)} > {config.max_response_bytes} bytes"
                )
            data = response.json()
            if not isinstance(data, dict):
                raise ValueError("External endpoint response must be a JSON object")
            return data
        except (httpx.HTTPError, ValueError) as exc:
            last_error = exc
            if attempt >= attempts:
                break
            logger.warning(
                "External endpoint request failed for %s: %s; retrying (%d/%d)",
                config.model_id,
                exc,
                attempt,
                attempts,
            )
            _sleep_before_retry(attempt)
    raise RuntimeError(f"External endpoint failed for {config.model_id}: {last_error}")


def _sleep_before_retry(attempt: int) -> None:
    time.sleep(min(8, 2 ** attempt))


def parse_external_forecast(
    payload: dict[str, Any],
    *,
    prediction_length: int,
    quantile_levels: list[float],
) -> np.ndarray:
    """Parse the community endpoint forecast response into GluonTS arrays."""

    outputs = payload.get("outputs") or payload.get("forecasts")
    if not isinstance(outputs, list) or not outputs:
        raise ValueError("External endpoint response must include a non-empty outputs list")
    first = outputs[0]
    if not isinstance(first, dict):
        raise ValueError("External endpoint output item must be an object")

    mean_values = first.get("mean")
    if mean_values is None:
        mean_values = first.get("prediction")
    if mean_values is None:
        mean_values = _lookup_quantile(first, 0.5)
    if mean_values is None:
        raise ValueError("External endpoint response is missing mean or median forecast")

    rows = [_forecast_vector(mean_values, prediction_length)]
    for level in quantile_levels:
        q_values = _lookup_quantile(first, level)
        rows.append(rows[0].copy() if q_values is None else _forecast_vector(q_values, prediction_length))
    forecast = np.stack(rows, axis=0).astype(np.float64)
    if not np.all(np.isfinite(forecast)):
        raise ValueError("External endpoint returned non-finite forecast values")
    return forecast


def _lookup_quantile(output: dict[str, Any], level: float) -> Any | None:
    for container_name in ("quantiles", "quantile_predictions"):
        container = output.get(container_name)
        if isinstance(container, dict):
            for key in (f"{level:g}", str(level), f"q{level:g}"):
                if key in container:
                    return container[key]
        elif isinstance(container, list):
            for item in container:
                if not isinstance(item, dict):
                    continue
                raw_level = item.get("level", item.get("quantile"))
                try:
                    item_level = float(raw_level)
                except (TypeError, ValueError):
                    continue
                if abs(item_level - level) < 1e-8:
                    return item.get("values")
    return None


def _forecast_vector(values: Any, prediction_length: int) -> np.ndarray:
    arr = np.asarray(values, dtype=np.float64)
    if arr.ndim == 2 and arr.shape[1] == 1:
        arr = arr[:, 0]
    elif arr.ndim > 1:
        arr = arr.reshape(arr.shape[0], -1).mean(axis=1)
    arr = arr.reshape(-1)
    if arr.size < prediction_length:
        raise ValueError(f"Forecast length {arr.size} < expected {prediction_length}")
    return arr[:prediction_length]