1"""Lipschitz-inflated split-conformal trajectory tube MVP."""
 2from dataclasses import dataclass
 3import numpy as np
 4
 5
 6def conformal_quantile(scores, alpha):
 7    """Finite-sample upper order statistic: rank ceil((m+1)(1-alpha))."""
 8    x = np.sort(np.asarray(scores, dtype=float))
 9    rank = int(np.ceil((len(x) + 1) * (1 - alpha)))
10    return float(x[min(max(rank - 1, 0), len(x) - 1)])
11
12
13def max_grid_slope(x, t):
14    return float(np.max(np.abs(np.diff(x) / np.diff(t))))
15
16@dataclass
17class TrajectoryTube:
18    q: float
19    gamma: float
20
21    @classmethod
22    def calibrate(cls, trajectories, obs_times, alpha=0.1, Lhat=None,
23                  predictor=lambda t, x: np.zeros_like(t), Lpred=0.0):
24        scores = []
25        for x, t in zip(trajectories, obs_times):
26            scores.append(np.max(np.abs(x - predictor(t, x))))
27        q = conformal_quantile(scores, alpha)
28        if Lhat is None:
29            raise ValueError("Lhat must be estimated on an independent high-frequency split")
30        return cls(q=q, gamma=float(Lhat + Lpred))
31
32    def radius(self, requested_times, observed_times):
33        delta = np.min(np.abs(np.asarray(requested_times)[:, None] -
34                              np.asarray(observed_times)[None, :]), axis=1)
35        return self.q + self.gamma * delta, delta
36
37    def contains(self, truth, center, requested_times, observed_times):
38        r, _ = self.radius(requested_times, observed_times)
39        return np.abs(np.asarray(truth) - np.asarray(center)) <= r + 1e-12