""" Walk-forward validation engine — the trust backbone. For a fixed config, roll the *same* ensemble forecast the dashboard makes across many recent origins, hold out the truth each time, and measure how good (and how honest) the forecast actually is on THIS ticker: * accuracy : MAPE / MAE / DirAcc of the ensemble median * skill : MASE and skill-vs-random-walk (does it beat "tomorrow = today"?) * calibration : the 10-90% band's real coverage, and a multiplicative band-scale that corrects it to a target coverage (split-conformal, CQR-style) * validity : mean range_ratio of the fed windows (Kronos normalisation sanity) Everything downstream (autoconfig, the trust card) is computed from this one pass so a config is judged by evidence, not by guesswork. """ import time import numpy as np # predict_us lives one level up when imported as `us.validation`, or flat in the Space. try: from us.predict_us import fetch_ohlcv, report_metrics, baseline_skill # type: ignore except Exception: # flat context (Space) or run from inside us/ from predict_us import fetch_ohlcv, report_metrics, baseline_skill # type: ignore _COLS = ["open", "high", "low", "close", "volume"] def make_windows(n_bars: int, lookback: int, pred_len: int, n_windows: int) -> list: """End-indices (exclusive) for rolling origins, spread across the recent slice.""" span = lookback + pred_len first = span + pred_len # leave room for one extra step back last = n_bars if first >= last: first = span return [int(x) for x in np.linspace(first, last, n_windows, dtype=int)] def _draw_paths(predictor, hist, fut_ts, pred_len, T, top_p, top_k, n_paths, seed): import torch paths = [] x_df = hist[_COLS].reset_index(drop=True) x_ts = hist["timestamps"].reset_index(drop=True) for i in range(n_paths): torch.manual_seed(seed + i); np.random.seed(seed + i) pred_df = predictor.predict( df=x_df, x_timestamp=x_ts, y_timestamp=fut_ts, pred_len=pred_len, T=T, top_k=top_k, top_p=top_p, sample_count=1, verbose=False, ) paths.append(pred_df["close"].values.astype(float)) return np.stack(paths, axis=0) # (n_paths, pred_len) def _best_band_scale(median, lo, hi, truth, target, grid=None): """Multiplicative scale s so [median±s·halfwidth] hits `target` coverage (CQR-style).""" if grid is None: grid = np.round(np.arange(0.3, 6.001, 0.05), 3) best_s, best_cov, best_gap = 1.0, float("nan"), 1e9 for s in grid: lo_s = median - s * (median - lo) hi_s = median + s * (hi - median) cov = float(np.mean((truth >= lo_s) & (truth <= hi_s))) gap = abs(cov - target) # prefer the tightest band whose coverage is closest to target if gap < best_gap - 1e-9 or (abs(gap - best_gap) <= 1e-9 and s < best_s): best_s, best_cov, best_gap = float(s), cov, gap return best_s, best_cov def _pearson(a, b): a = np.asarray(a, float); b = np.asarray(b, float) if len(a) < 3 or np.std(a) < 1e-12 or np.std(b) < 1e-12: return float("nan") return float(np.corrcoef(a, b)[0, 1]) def compute_ic(records): """Information Coefficient: correlation of predicted vs realised step-returns, pooled across windows. This is how the Kronos paper judges skill (direction/ranking is what a generative price model can actually capture; absolute price level is near-random). Returns (IC=Pearson, RankIC=Spearman).""" pr, tr = [], [] for w in records: if len(w["median"]) > 1: pr.append(np.diff(w["median"])); tr.append(np.diff(w["truth"])) if not pr: return float("nan"), float("nan") pr = np.concatenate(pr); tr = np.concatenate(tr) ic = _pearson(pr, tr) rank = lambda x: np.argsort(np.argsort(x)).astype(float) ric = _pearson(rank(pr), rank(tr)) return ic, ric def run_windows(predictor, df, lookback, pred_len, T=0.7, top_p=0.9, top_k=0, n_paths=10, windows=None, n_windows=6, seed=123): """Draw the ensemble at each rolling origin; return a per-window record list (raw arrays + point metrics). Explicit `windows` lets callers share identical origins across configs (needed for a fair nested inner/outer split).""" if windows is None: windows = make_windows(len(df), lookback, pred_len, n_windows) records = [] for end in windows: hist = df.iloc[end - lookback - pred_len:end - pred_len] fut = df.iloc[end - pred_len:end] if len(hist) < lookback or len(fut) < pred_len: continue fut_ts = fut["timestamps"].reset_index(drop=True) paths = _draw_paths(predictor, hist, fut_ts, pred_len, T, top_p, top_k, n_paths, seed) median = np.median(paths, axis=0) lo = np.quantile(paths, 0.1, axis=0) hi = np.quantile(paths, 0.9, axis=0) truth = fut["close"].values.astype(float) hist_close = hist["close"].values.astype(float) wmin, wmax = float(hist_close.min()), float(hist_close.max()) rec = {"median": median, "lo": lo, "hi": hi, "truth": truth, "range_ratio": wmax / wmin if wmin > 0 else float("inf")} rec.update(report_metrics(truth, median)) rec.update(baseline_skill(hist_close, truth, median)) records.append(rec) return records def aggregate(records, target_coverage=0.8, band_scale=None) -> dict: """Summarise a set of window records. If band_scale is None, FIT the conformal scale to hit target coverage on these records (calibration set). If a band_scale is given, APPLY it and just measure the resulting coverage (test set) — this is how the nested outer set reports coverage using a scale fitted only on the inner set.""" if not records: raise ValueError("No valid windows — not enough history for this lookback/pred_len.") keys = ["MAE", "RMSE", "MAPE%", "DirAcc%", "MASE", "skill_vs_rw"] agg = {k: float(np.nanmean([w[k] for w in records])) for k in keys} ic, ric = compute_ic(records) agg["IC"] = ic; agg["RankIC"] = ric med = np.concatenate([w["median"] for w in records]) lo = np.concatenate([w["lo"] for w in records]) hi = np.concatenate([w["hi"] for w in records]) tr = np.concatenate([w["truth"] for w in records]) cov_raw = float(np.mean((tr >= lo) & (tr <= hi))) if band_scale is None: bs, cov_cal = _best_band_scale(med, lo, hi, tr, target_coverage) else: bs = float(band_scale) lo_s = med - bs * (med - lo); hi_s = med + bs * (hi - med) cov_cal = float(np.mean((tr >= lo_s) & (tr <= hi_s))) rr = [w["range_ratio"] for w in records] return { "n_windows": len(records), "metrics": {k: (round(v, 4) if v == v else None) for k, v in agg.items()}, "calibration": {"target_coverage": target_coverage, "coverage_raw": round(cov_raw, 4), "band_scale": round(float(bs), 3), "coverage_calibrated": round(cov_cal, 4)}, "validity": {"range_ratio_mean": round(float(np.mean(rr)), 3), "range_ratio_max": round(float(np.max(rr)), 3)}, } def evaluate(predictor, df, lookback, pred_len, T=0.7, top_p=0.9, top_k=0, n_paths=10, n_windows=6, seed=123, target_coverage=0.8) -> dict: """Single-pass walk-forward for one config (non-nested; used for quick checks).""" t0 = time.time() records = run_windows(predictor, df, lookback, pred_len, T=T, top_p=top_p, top_k=top_k, n_paths=n_paths, n_windows=n_windows, seed=seed) out = aggregate(records, target_coverage) out["config"] = {"lookback": lookback, "pred_len": pred_len, "T": T, "top_p": top_p, "top_k": top_k, "n_paths": n_paths} out["secs"] = round(time.time() - t0, 1) return out