tinycast/predictor.py

"""GIFT-Eval gluonts predictor for TinyCast.

``TinyCastPredictor`` is the deployed predictor. It wraps the model in the
gluonts Predictor protocol and drives autoregressive-rollout decoding
(48-step chunks), flip-invariance symmetrization, NaN-imputation and
optional period-alignment downsampling.

Anything that changes what the predictor emits says so:

``device``          a named device must exist. ``device=None`` is the only
                    request that selects one for you (CUDA when present, else
                    CPU); the resolved device is ``predictor.device``.
``TINYCAST_INT8``   ``w8`` or ``w8a8`` post-training fake quantization, off by
                    default. Any other non-empty value is an error, not an
                    inert setting.
``TINYCAST_TILT_K`` an eval-only probe that moves the emitted median off the
                    quantile grid, off (``0``) by default, with
                    ``TINYCAST_TILT_MODE`` in {``adaptive``, ``fixed``}.

The environment variables are read once, when the predictor is constructed, and
an engaged one prints a line naming itself. Reproducing the published numbers
means leaving them unset.
"""
from __future__ import annotations

import math
import os
from typing import List, Optional, Tuple

import numpy as np
import torch
import torch.nn as nn

from .checkpoint import load_checkpoint
from .normalization import WindowMinMax
from .scale import seasonal_scale_factor

try:
    from torch.amp import autocast as _autocast_fp
except Exception:  # pragma: no cover
    _autocast_fp = None


def _resolve_device(requested: Optional[str]) -> torch.device:
    """Return the requested device, or say why it is unavailable.

    ``None`` is the request to choose: CUDA when it is available, CPU otherwise.
    Every other value is a requirement. Falling back to CPU behind the caller's
    back drops bf16 autocast for fp32 and moves every forecast by roughly 1e-3,
    which is small enough to read as a regression and large enough to fail an
    assertion against the published aggregates, so an absent accelerator is an
    error rather than a substitution.
    """
    if requested is None:
        return torch.device("cuda" if torch.cuda.is_available() else "cpu")
    dev = torch.device(requested)
    if dev.type == "cuda":
        if not torch.cuda.is_available():
            raise RuntimeError(
                f"device={requested!r} was requested but "
                "torch.cuda.is_available() is False. Pass device='cpu' to run "
                "on CPU (fp32, and about 1e-3 off the published numbers), or "
                "device=None to take whichever device is present."
            )
        if dev.index is not None and dev.index >= torch.cuda.device_count():
            raise RuntimeError(
                f"device={requested!r} was requested but only "
                f"{torch.cuda.device_count()} CUDA device(s) are visible."
            )
    if dev.type == "mps" and not torch.backends.mps.is_available():
        raise RuntimeError(
            f"device={requested!r} was requested but MPS is unavailable. "
            "Pass device='cpu' to run on CPU."
        )
    return dev


_INT8_MODES = ("w8", "w8a8")
_INT8_OFF = ("", "0", "off", "no", "false", "none")
_TILT_MODES = ("adaptive", "fixed")


def _resolve_int8() -> Optional[str]:
    """Read ``TINYCAST_INT8``: the quantization mode, or ``None`` when off.

    Only ``w8`` and ``w8a8`` name a scheme. A truthy value such as ``1`` names
    no scheme, and picking one for the caller would quantize a run that asked
    for something else, so an unrecognized value raises. The alternative is what
    this used to do: leave the run in fp32 while the caller believes they are
    measuring INT8.
    """
    raw = os.environ.get("TINYCAST_INT8", "").strip().lower()
    if raw in _INT8_OFF:
        return None
    if raw not in _INT8_MODES:
        raise ValueError(
            f"TINYCAST_INT8={raw!r} is not a quantization mode. Use 'w8' "
            "(per-channel INT8 weights) or 'w8a8' (+ per-tensor dynamic INT8 "
            "activations), or unset it to run in floating point."
        )
    return raw


def _resolve_tilt() -> Tuple[float, str]:
    """Read the tilt probe's variables, announcing an engaged one.

    ``TINYCAST_INT8`` prints a line when it engages; this does the same, so
    neither switch can invalidate a golden array without appearing in the log.
    ``TINYCAST_TILT_MODE`` is checked only when the tilt is on, which keeps it
    inert at ``K=0`` as documented.
    """
    raw = os.environ.get("TINYCAST_TILT_K", "").strip()
    try:
        k = float(raw) if raw else 0.0
    except ValueError:
        raise ValueError(
            f"TINYCAST_TILT_K={raw!r} is not a number. Unset it, or set 0, to "
            "emit the model's own median."
        ) from None
    mode = os.environ.get("TINYCAST_TILT_MODE", "adaptive").strip().lower()
    if k != 0.0:
        if mode not in _TILT_MODES:
            raise ValueError(
                f"TINYCAST_TILT_MODE={mode!r} is not a tilt rule. Use "
                "'adaptive' (skew-following) or 'fixed'."
            )
        print(
            f"[tilt] TINYCAST_TILT_K={k:g} {mode}: the emitted median is "
            "re-interpolated off the quantile grid, so this is not the "
            "published model's forecast (inert below 3 quantiles)",
            flush=True,
        )
    return k, mode


def _numpy_fill(arr: np.ndarray) -> np.ndarray:
    mask = np.isnan(arr)
    idx = np.where(~mask, np.arange(mask.shape[1]), 0)
    np.maximum.accumulate(idx, axis=1, out=idx)
    return arr[np.arange(idx.shape[0])[:, None], idx]


class ARRolloutPredictor:
    """Base predictor: AR rollout + flip + NaN-imputation + period downsample.

    Subclasses install ``self.model`` (a callable ``(x, x_mark, y_mark)``).
    ``device=None`` selects one; a named device must exist. Either way the
    device that ran is ``self.device``.

    Every setting that changes a forecast is a named parameter, so a keyword
    that is not one raises. Absorbing unknown keywords would let a misspelling
    such as ``force_flip_invarience=True`` construct a predictor with flip
    symmetrization off and return a forecast that looks right and cannot be
    reproduced.
    """

    def __init__(
        self,
        prediction_length: int,
        device: Optional[str] = None,
        seq_len: int = 2048,
        input_token_len: int = 2048,
        output_token_len: int = 48,
        num_samples: int = 100,
        batch_size: int = 256,
        use_amp: int = 1,
        downsample_factor: int = 1,
        force_flip_invariance: bool = False,
        adaptive_ar_rollout: bool = False,
    ):
        self.device = _resolve_device(device)
        self.prediction_length = int(prediction_length)
        self.num_samples = int(num_samples)
        self.batch_size = int(batch_size)
        self.seq_len = int(seq_len)
        self.input_token_len = int(input_token_len)
        self.output_token_len = int(output_token_len)
        self.use_amp = int(use_amp)
        self.downsample_factor = int(downsample_factor)
        self.force_flip_invariance = bool(force_flip_invariance)
        self.adaptive_ar_rollout = bool(adaptive_ar_rollout)
        # Quantile levels emitted by the model (set by subclasses from cfg).
        self.quantiles = [0.5]
        self.model = None

    def _downsample_if_needed(
        self, series: torch.Tensor
    ) -> Tuple[torch.Tensor, int]:
        cur = series
        if self.downsample_factor > 1:
            cur = cur[::self.downsample_factor]
        return cur, self.downsample_factor

    def _left_pad_to_len(
        self, arr: np.ndarray, target_len: int
    ) -> Tuple[np.ndarray, int]:
        if arr.shape[0] >= target_len:
            return arr[-target_len:], 0
        pad_len = target_len - arr.shape[0]
        fill_value = arr[0] if arr.shape[0] > 0 else 0.0
        padding = np.full((pad_len,), fill_value, dtype=arr.dtype)
        return np.concatenate([padding, arr], axis=0), pad_len

    def _prepare_context_matrix(
        self, context: List[torch.Tensor]
    ) -> Tuple[torch.Tensor, List[int]]:
        xs = []
        downsample_factors = []
        for c in context:
            cur, df = self._downsample_if_needed(c)
            downsample_factors.append(df)

            cur_np = cur.detach().cpu().float().numpy()
            cur_np, _ = self._left_pad_to_len(cur_np, self.seq_len)

            x2d = cur_np[None, :]
            x_interp = np.copy(x2d)
            series = x2d[0]
            if np.any(np.isnan(series)):
                valid_mask = ~np.isnan(series)
                if np.sum(valid_mask) >= 2:
                    valid_idx = np.where(valid_mask)[0]
                    valid_val = series[valid_mask]
                    x_interp[0] = np.interp(
                        np.arange(len(series)), valid_idx, valid_val
                    )
                else:
                    x_interp = _numpy_fill(x2d)
            ff = _numpy_fill(x_interp)
            bf = np.flip(_numpy_fill(np.flip(x_interp, axis=1)), axis=1)
            x_imp = np.where(np.isnan(ff), bf, ff)
            x_imp = np.where(np.isnan(x_imp), 0.0, x_imp)
            xs.append(x_imp[0])

        x = torch.tensor(
            np.stack(xs), device=self.device, dtype=torch.float32
        ).unsqueeze(-1)
        return x, downsample_factors

    def _decode_autoregressive(
        self,
        init_ctx: torch.Tensor,
        use_bf16: bool,
        downsample_factors: List[int],
    ) -> torch.Tensor:
        B, _, C = init_ctx.shape
        roll_len = int(self.output_token_len)

        if self.adaptive_ar_rollout:
            try:
                from .periodogram import significant_periods
                periods, _s, _nv = significant_periods(
                    init_ctx[:, -self.seq_len:, 0].float(),
                    min_period=2, max_period=self.seq_len // 2, top_k=1,
                )
                pos = periods[periods > 0]
                if pos.numel() > 0:
                    p0 = int(pos.float().median().item())
                    if 2 <= p0 < roll_len:
                        roll_len = max(4, p0)
            except Exception:
                pass

        target_pred_lens = [
            int(self.prediction_length) // int(max(1, df))
            for df in downsample_factors
        ]
        max_target_pred_len = max(target_pred_lens)
        steps = math.ceil(max_target_pred_len / roll_len)
        preds: List[torch.Tensor] = []
        batch_ctx = init_ctx

        y_mark = torch.zeros(
            B, self.output_token_len, C,
            device=self.device, dtype=init_ctx.dtype,
        )

        for _ in range(steps):
            x_in = batch_ctx[:, -self.seq_len:, :]
            x_mark = torch.zeros_like(x_in)
            if _autocast_fp is not None and self.use_amp and use_bf16:
                try:
                    with _autocast_fp("cuda", dtype=torch.bfloat16):
                        out = self.model(x_in, x_mark, y_mark)
                except Exception:
                    out = self.model(x_in, x_mark, y_mark)
            else:
                out = self.model(x_in, x_mark, y_mark)
            chunk = out[:, -self.output_token_len:, :][:, :roll_len, :]  # (B, roll, Q)
            preds.append(chunk)
            # Feed only the MEDIAN quantile back into the context (the AR state
            # is a univariate series); keep all Q in preds for the forecast.
            q_mid = chunk.shape[-1] // 2
            batch_ctx = torch.cat([batch_ctx, chunk[:, :, q_mid:q_mid + 1]], dim=1)

        return torch.cat(preds, dim=1)                                   # (B, pl, Q)

    @torch.no_grad()
    def predict(self, test_data_input, use_bf16_if_available: bool = True):
        from gluonts.itertools import batcher
        from gluonts.model.forecast import SampleForecast, QuantileForecast

        forecasts: List = []
        use_bf16 = bool(
            use_bf16_if_available
            and self.device.type == "cuda"
            and torch.cuda.is_available()
            and torch.cuda.is_bf16_supported()
        )

        for batch in batcher(test_data_input, batch_size=self.batch_size):
            targets = [
                torch.tensor(entry["target"], dtype=torch.float32)
                for entry in batch
            ]
            batch_ctx, dfs = self._prepare_context_matrix(targets)
            pred_pos = self._decode_autoregressive(batch_ctx, use_bf16, dfs)  # (B,pl,Q)
            if self.force_flip_invariance:
                pred_neg = self._decode_autoregressive(-batch_ctx, use_bf16, dfs)
                # Flip-symmetrize. For quantiles the tau-quantile of -y is
                # -(the (1-tau)-quantile of y), so reverse the quantile axis on
                # the negated branch. Q=1 (median) reverse is a no-op.
                pred = 0.5 * (pred_pos - pred_neg.flip(dims=[-1]))
            else:
                pred = pred_pos

            Q = pred.shape[-1]
            pred_np = pred.float().detach().cpu().numpy()                 # (B, pl, Q)
            if not np.isfinite(pred_np).all():
                for qi in range(Q):
                    pred_np[:, :, qi] = _numpy_fill(pred_np[:, :, qi])

            for i, ts in enumerate(batch):
                df = int(max(1, dfs[i]))
                target_pl = int(self.prediction_length) // df
                arr = pred_np[i, :target_pl, :]                          # (target_pl, Q)
                if df > 1:
                    new_len = int(self.prediction_length)
                    src = np.linspace(0, 1, arr.shape[0])
                    dst = np.linspace(0, 1, new_len)
                    arr = np.stack([np.interp(dst, src, arr[:, qi])
                                    for qi in range(Q)], axis=1)         # (new_len, Q)
                start_date = ts["start"] + len(ts["target"])
                if Q > 1:
                    # Sort across quantiles to guarantee non-crossing, then a
                    # QuantileForecast so gluonts scores WQL/CRPS over the deciles.
                    arr = np.sort(arr, axis=1)
                    forecasts.append(QuantileForecast(
                        forecast_arrays=arr.T,                          # (Q, pl)
                        start_date=start_date,
                        forecast_keys=[str(q) for q in self.quantiles],
                    ))
                else:
                    samples = np.repeat(arr[:, 0][None, :], self.num_samples, axis=0)
                    forecasts.append(
                        SampleForecast(samples=samples, start_date=start_date)
                    )

        return forecasts


class _BackboneAdapter(nn.Module):
    """Adapts the model wrapper to the ``(x, x_mark, y_mark) -> (B, p, Q)`` contract."""

    def __init__(self, backbone: nn.Module, scale_factor: float = 1.0):
        super().__init__()
        self.backbone = backbone
        self.scale_factor = float(scale_factor)
        # single-shot counterfactual: emit this many steps in ONE forward.
        self.pred_len_override = None
        # Resolved once, and announced when on: a forecast must not change
        # because a variable was exported after the predictor was built.
        self.tilt_k, self.tilt_mode = _resolve_tilt()

    def forward(self, x, x_mark=None, y_mark=None, **kwargs):
        y_norm, x_min, x_range = self.backbone.encode(
            x, batch_first=True, scale_factor=self.scale_factor,
            horizon=self.pred_len_override,
        )
        # Match training: the chunk loss clamps y_norm to [-5,5] before the
        # pinball loss, so the model is never optimized outside that band.
        # Clamp at inference too: otherwise an un-penalized overshoot feeds
        # back into the AR rollout context and compounds over chunks.
        y_norm = y_norm.clamp(-5.0, 5.0)

        # num_quantiles > 1 => y_norm is (B, p, Q). Manual inverse since
        # WindowMinMax.inverse_transform squeezes x_min/x_range for the (B, p)
        # point case; x_min/x_range are (B, 1, 1) so they broadcast directly.
        if y_norm.dim() == 3:
            tk = self.tilt_k
            Q = y_norm.shape[-1]
            if tk != 0.0 and Q >= 3:
                # EVAL-ONLY de-hedge probe (no training); default off (tk=0).
                qm = Q // 2
                lo = y_norm[..., 0]; hi = y_norm[..., -1]; med = y_norm[..., qm]
                if self.tilt_mode == "fixed":
                    tau = torch.full_like(med, 0.5 + tk)
                else:                                            # skew-adaptive
                    asym = (hi + lo - 2 * med) / (hi - lo).abs().clamp(min=1e-6)
                    tau = 0.5 + tk * torch.tanh(asym)            # >0 right-skew
                tau = tau.clamp(qm / (Q + 1.0), (qm + 2) / (Q + 1.0))
                idxf = (tau * (Q + 1) - 1).clamp(0, Q - 1 - 1e-4)
                ilo = idxf.floor().long().clamp(0, Q - 2)
                frac = (idxf - ilo.to(idxf.dtype)).clamp(0, 1)
                qlo = torch.gather(y_norm, -1, ilo.unsqueeze(-1)).squeeze(-1)
                qhi = torch.gather(y_norm, -1, (ilo + 1).unsqueeze(-1)).squeeze(-1)
                y_norm = y_norm.clone()
                y_norm[..., qm] = qlo * (1 - frac) + qhi * frac
            return y_norm * x_range + x_min                    # (B, p, Q)
        y_pred = WindowMinMax.inverse_transform(y_norm, x_min, x_range)
        return y_pred.unsqueeze(-1)  # (B, p, 1)


class TinyCastPredictor(ARRolloutPredictor):
    """The deployed predictor: AR-rollout decoding around the TinyCast model."""

    def __init__(
        self,
        prediction_length: int,
        checkpoint_path: str,
        device: Optional[str] = None,
        num_samples: int = 100,
        batch_size: int = 256,
        use_amp: int = 1,
        downsample_factor: int = 1,
        force_flip_invariance: bool = False,
        freq: Optional[str] = None,
        domain: Optional[str] = None,
        no_daily: bool = False,
        single_shot: bool = False,
        adaptive_ar_rollout: bool = False,
        config_path: Optional[str] = None,
    ):
        self.single_shot = bool(single_shot)
        # 1. Load the weights (safetensors + config.json).
        model, cfg = load_checkpoint(checkpoint_path, config_path)

        # 2. Eval-time scale_factor from (freq, domain). bizitobs_l2c has no
        #    daily cycle -> /7.
        if freq is not None and domain is not None:
            sf_eval = seasonal_scale_factor(freq, domain)
            if no_daily:
                sf_eval /= 7
        else:
            sf_eval = 1.0
        self._eval_scale_factor = float(sf_eval)

        # 3. Base predictor bookkeeping (drives AR-rollout / batching / flip).
        super().__init__(
            prediction_length=prediction_length,
            device=device,
            seq_len=int(cfg.seq_len),
            input_token_len=int(cfg.seq_len),
            output_token_len=int(cfg.output_token_len),
            num_samples=num_samples,
            batch_size=batch_size,
            use_amp=use_amp,
            downsample_factor=downsample_factor,
            force_flip_invariance=force_flip_invariance,
            adaptive_ar_rollout=adaptive_ar_rollout,
        )
        self.quantiles = list(getattr(cfg, "quantiles", [0.5])) or [0.5]
        self._missing_aware = bool(getattr(cfg, "missing_channel", False))

        # 4. Install the model wrapped in the adapter interface.
        backbone = model.model  # TinyCastForPrediction -> TinyCastBackbone
        backbone.to(self.device).eval()
        # Optional INT8 post-training fake-quant. Off by default;
        # TINYCAST_INT8=w8 (weights) or w8a8 (+ dynamic activations).
        # quantize_int8_ prints the mode it applied.
        _int8 = _resolve_int8()
        if _int8 is not None:
            from .quant import quantize_int8_
            quantize_int8_(backbone, _int8)
        self.model = _BackboneAdapter(
            backbone, scale_factor=self._eval_scale_factor,
        ).to(self.device).eval()
        if self.single_shot:
            self.model.pred_len_override = int(prediction_length)

    def _decode_autoregressive(self, init_ctx, use_bf16, downsample_factors):
        """Single-shot override: one forward for the full horizon (no AR loop).
        Default deployed behavior (single_shot=False) falls back to the base
        AR rollout."""
        if not self.single_shot:
            return super()._decode_autoregressive(init_ctx, use_bf16, downsample_factors)
        x_in = init_ctx[:, -self.seq_len:, :]
        x_mark = torch.zeros_like(x_in)
        use_cuda = bool(self.use_amp and use_bf16 and str(self.device).startswith("cuda"))
        if use_cuda:
            with torch.autocast("cuda", dtype=torch.bfloat16):
                out = self.model(x_in, x_mark, None)
        else:
            out = self.model(x_in, x_mark, None)
        return out  # (B, prediction_length, 1)

    def _prepare_context_matrix(self, context):
        """Missing-aware override: keep genuine gaps as NaN so the observed-mask
        + observed-only normalization see true missingness. Falls back to the
        base (interpolating) behavior when the model is not missing-aware
        (the deployed model is not)."""
        if not getattr(self, "_missing_aware", False):
            return super()._prepare_context_matrix(context)
        xs, dfs = [], []
        for c in context:
            cur, df = self._downsample_if_needed(c)
            dfs.append(df)
            a = cur.detach().cpu().float().numpy()
            if a.shape[0] >= self.seq_len:
                a = a[-self.seq_len:]
            else:
                pad = np.full((self.seq_len - a.shape[0],), np.nan, dtype=a.dtype)
                a = np.concatenate([pad, a], axis=0)
            xs.append(a)
        x = torch.tensor(
            np.stack(xs), device=self.device, dtype=torch.float32
        ).unsqueeze(-1)
        return x, dfs