"""aurora-blend-v2 — score-tuned long-context mixture with weather slice. Cascade score = geomean(CRPS, MASE) on held-out windows (lower is better). This prior is tuned so Toto2 learns *forecastable* multi-scale structure: * weather mass ~0.42 (king-92 skill signal: diurnal / radiation / cloud) * remaining mass on long-range AR / integrated / GP / seasonal (king-zenfro) * calendar-biased periods (24 / 168 / 7 / 12 / 720), higher SNR, quieter innov * weekly-structured intermittent demand for retail-like domains Hot recurrences stay numba-jitted (cache=False for cascade's dynamic loader). Chunked emission + fixed L=4096 keep generation wall-safe under the token wall. """ from __future__ import annotations import json from collections.abc import Callable, Iterator from pathlib import Path from typing import Any import numpy as np from numba import njit from cascade.interface import DataGenerator _BATCH = 512 # Block size for the segmented AR(1) scan (~2*sqrt(L) iters instead of L). _AR1_BLOCK = 32 _PRIOR_NAMES: tuple[str, ...] = ( "trend_seasonal_ar", "regime_shift", "multiplicative", "ar2", "integrated", "threshold_ar", "chaotic", "rff_gp", "intermittent", "pulse_outlier", "weather", ) # Defaults mirror config.json; config overrides win. _PRIOR_MASS: dict[str, float] = { "weather": 0.42, "trend_seasonal_ar": 0.15, "regime_shift": 0.08, "ar2": 0.09, "integrated": 0.09, "rff_gp": 0.07, "multiplicative": 0.05, "threshold_ar": 0.02, "chaotic": 0.01, "intermittent": 0.015, "pulse_outlier": 0.005, } # Period menu + sampling weights: bias toward cadences that dominate real eval # (hourly diurnal, weekly, daily, monthly) while retaining long-context options. _SEASON_PERIODS = np.asarray( [4.0, 7.0, 12.0, 24.0, 30.0, 52.0, 96.0, 144.0, 168.0, 336.0, 504.0, 672.0, 720.0], dtype=np.float64, ) _SEASON_P = np.asarray( [0.02, 0.12, 0.10, 0.22, 0.04, 0.04, 0.03, 0.03, 0.18, 0.05, 0.04, 0.05, 0.08], dtype=np.float64, ) _SEASON_P = _SEASON_P / _SEASON_P.sum() # ── numba kernels (cache=False: cascade imports this file as a dynamic module) ─ @njit(cache=False) def _jit_ar2(innov: np.ndarray, a1: np.ndarray, a2: np.ndarray) -> np.ndarray: n, L = innov.shape out = np.empty((n, L), dtype=np.float64) for i in range(n): aa1, aa2 = a1[i], a2[i] out[i, 0] = innov[i, 0] if L > 1: out[i, 1] = aa1 * out[i, 0] + innov[i, 1] for t in range(2, L): out[i, t] = aa1 * out[i, t - 1] + aa2 * out[i, t - 2] + innov[i, t] return out @njit(cache=False) def _jit_setar( innov: np.ndarray, phi_pos: np.ndarray, phi_neg: np.ndarray, c_pos: np.ndarray, c_neg: np.ndarray, ) -> np.ndarray: n, L = innov.shape out = np.empty((n, L), dtype=np.float64) for i in range(n): out[i, 0] = innov[i, 0] for t in range(1, L): prev = out[i, t - 1] if prev >= 0.0: v = c_pos[i] + phi_pos[i] * prev + innov[i, t] else: v = c_neg[i] + phi_neg[i] * prev + innov[i, t] if v > 1e6: v = 1e6 elif v < -1e6: v = -1e6 out[i, t] = v return out @njit(cache=False) def _jit_chaos( pick_sine: np.ndarray, r_log: np.ndarray, r_sin: np.ndarray, x0: np.ndarray, L: int, ) -> np.ndarray: n = x0.shape[0] out = np.empty((n, L), dtype=np.float64) for i in range(n): cur = x0[i] out[i, 0] = cur sine = pick_sine[i] rl, rs = r_log[i], r_sin[i] for t in range(1, L): if sine: cur = rs * np.sin(np.pi * cur) else: cur = rl * cur * (1.0 - cur) if cur < 0.0: cur = 0.0 elif cur > 1.0: cur = 1.0 out[i, t] = cur return out @njit(cache=False) def _jit_holds(series: np.ndarray, hold: np.ndarray) -> None: n, L = series.shape for i in range(n): for t in range(1, L): if hold[i, t]: series[i, t] = series[i, t - 1] # ── thin numpy wrappers ───────────────────────────────────────────────────── def _ar1(innov: np.ndarray, phi: np.ndarray, S: int = _AR1_BLOCK) -> np.ndarray: """AR(1) via segmented (block) scan — ~2*sqrt(L) Python iters, same law as loop. Weather/cloud call this many times per batch; a pure-numpy scan beats a per-row numba loop here and feeds the trainer more tokens before the wall. """ n, L = innov.shape p = np.asarray(phi, dtype=np.float64).reshape(n) if L < 2 * S: x = np.empty((n, L), dtype=np.float64) x[:, 0] = innov[:, 0] for t in range(1, L): x[:, t] = p * x[:, t - 1] + innov[:, t] return x B = L // S body = B * S main = innov[:, :body].reshape(n, B, S) y = np.empty((n, B, S), dtype=np.float64) y[:, :, 0] = main[:, :, 0] pcol = p[:, None] for s in range(1, S): y[:, :, s] = pcol * y[:, :, s - 1] + main[:, :, s] r = p ** S ylast = y[:, :, S - 1] X = np.empty((n, B), dtype=np.float64) X[:, 0] = ylast[:, 0] for b in range(1, B): X[:, b] = r * X[:, b - 1] + ylast[:, b] carry_in = np.empty((n, B), dtype=np.float64) carry_in[:, 0] = 0.0 carry_in[:, 1:] = X[:, :-1] ppow = p[:, None] ** np.arange(1, S + 1, dtype=np.float64)[None, :] x = y + carry_in[:, :, None] * ppow[:, None, :] x = x.reshape(n, body) if body == L: return x out = np.empty((n, L), dtype=np.float64) out[:, :body] = x prev = out[:, body - 1] for t in range(body, L): prev = p * prev + innov[:, t] out[:, t] = prev return out def _ar2(innov: np.ndarray, a1: np.ndarray, a2: np.ndarray) -> np.ndarray: return _jit_ar2( np.ascontiguousarray(innov), np.ascontiguousarray(a1.reshape(-1)), np.ascontiguousarray(a2.reshape(-1)), ) def _pick_periods(rng: np.random.Generator, n: int) -> np.ndarray: return rng.choice(_SEASON_PERIODS, size=n, p=_SEASON_P)[:, None] def _harmonics(rng: np.random.Generator, n: int, L: int, *, depth: int = 3) -> np.ndarray: """Sum of up to ``depth`` sinusoids; periods biased to calendar cadences.""" t = np.arange(L, dtype=np.float64)[None, :] n_comp = rng.integers(1, depth + 1, size=n) acc = np.zeros((n, L), dtype=np.float64) for j in range(depth): on = (n_comp > j).astype(np.float64)[:, None] period = _pick_periods(rng, n) # Slightly stronger amps on the first component (dominant seasonal). lo, hi = (0.4, 2.2) if j == 0 else (0.15, 1.2) amp = rng.uniform(lo, hi, size=n)[:, None] phase = rng.uniform(0.0, 2.0 * np.pi, size=n)[:, None] acc += on * amp * np.sin(2.0 * np.pi * t / period + phase) return acc def _jumps(rng: np.random.Generator, n: int, L: int, rate: float, scale) -> np.ndarray: hit = rng.random((n, L)) < rate mag = rng.normal(0.0, 1.0, size=(n, L)) s = np.asarray(scale, dtype=np.float64) if s.ndim == 1: s = s[:, None] out = hit * mag * s out[:, 0] = 0.0 return out def _finite(block: np.ndarray) -> np.ndarray: x = np.asarray(block, dtype=np.float64) x = np.nan_to_num(x, nan=0.0, posinf=1e6, neginf=-1e6) return np.clip(x, -1e6, 1e6) # ── family emitters ───────────────────────────────────────────────────────── def emit_trend_seasonal( rng: np.random.Generator, n: int, L: int, *, hi_frac: float, exc_lo: float, exc_hi: float, ) -> np.ndarray: t = np.arange(L, dtype=np.float64)[None, :] level = rng.normal(0.0, 1.0, size=(n, 1)) heavy = rng.random((n, 1)) < hi_frac slope = np.where( heavy, rng.normal(0.0, exc_hi, size=(n, 1)), rng.normal(0.0, exc_lo, size=(n, 1)), ) tn = t / max(L - 1, 1) signal = level + slope * tn + _harmonics(rng, n, L, depth=3) # More persistent AR noise, lower sigma → higher SNR / easier forecasts phi = rng.uniform(0.35, 0.92, size=n) sigma = rng.uniform(0.06, 0.40, size=(n, 1)) innov = rng.normal(0.0, 1.0, size=(n, L)) * sigma return signal + _ar1(innov, phi) def emit_regime(rng: np.random.Generator, n: int, L: int) -> np.ndarray: # Slightly fewer breaks than classic (2/L): long stable regimes are learnable. level = np.cumsum(_jumps(rng, n, L, rate=2.0 / L, scale=2.0), axis=1) log_vol = np.cumsum(_jumps(rng, n, L, rate=2.0 / L, scale=0.4), axis=1) vol = np.exp(np.clip(log_vol, -3.0, 3.0)) * rng.uniform(0.08, 0.40, size=(n, 1)) noise = rng.normal(0.0, 1.0, size=(n, L)) * vol seas = _harmonics(rng, n, L, depth=2) * rng.uniform(0.2, 1.0, size=(n, 1)) return level + seas + noise def emit_multiplicative( rng: np.random.Generator, n: int, L: int, *, hi_frac: float, exc_lo: float, exc_hi: float, ) -> np.ndarray: t = np.arange(L, dtype=np.float64)[None, :] heavy = rng.random((n, 1)) < hi_frac g = np.where( heavy, rng.normal(0.0, exc_hi, size=(n, 1)), rng.normal(0.0, exc_lo, size=(n, 1)), ) tn = t / max(L - 1, 1) base = np.exp(g * tn + rng.normal(0.0, 0.25, size=(n, 1))) amp = rng.uniform(0.15, 0.65, size=(n, 1)) per = rng.choice(np.array([7.0, 12.0, 24.0, 52.0, 168.0, 720.0]), size=n)[:, None] seas = 1.0 + amp * np.sin(2.0 * np.pi * t / per + rng.uniform(0.0, 2.0 * np.pi, size=(n, 1))) # Quieter multiplicative noise noise = 1.0 + rng.normal(0.0, 1.0, size=(n, L)) * rng.uniform(0.015, 0.10, size=(n, 1)) scale = rng.uniform(1.0, 50.0, size=(n, 1)) return scale * base * np.clip(seas, 0.05, None) * np.clip(noise, 0.05, None) def emit_ar2(rng: np.random.Generator, n: int, L: int) -> np.ndarray: # Bias toward near-unit-root persistence (finance / sensor domains). p1 = rng.uniform(0.45, 0.985, size=n) p2 = rng.uniform(-0.45, 0.45, size=n) a1 = p1 * (1.0 - p2) a2 = p2 sigma = rng.uniform(0.15, 0.55, size=(n, 1)) innov = rng.normal(0.0, 1.0, size=(n, L)) * sigma path = _ar2(innov, a1, a2) drift = rng.normal(0.0, 0.004, size=(n, 1)) * np.arange(L, dtype=np.float64)[None, :] return path + drift def emit_integrated(rng: np.random.Generator, n: int, L: int) -> np.ndarray: twice = rng.random(n) < 0.30 drift = rng.normal(0.0, 0.015, size=(n, 1)) sigma = rng.uniform(0.15, 0.75, size=(n, 1)) steps = rng.normal(0.0, 1.0, size=(n, L)) * sigma + drift walk = np.cumsum(steps, axis=1) walk2 = np.cumsum(walk, axis=1) return np.where(twice[:, None], walk2 / max(L, 1) ** 0.5, walk) def emit_threshold(rng: np.random.Generator, n: int, L: int) -> np.ndarray: phi_pos = rng.uniform(0.4, 0.92, size=n) phi_neg = rng.uniform(-0.7, 0.35, size=n) c_pos = rng.normal(0.0, 0.25, size=n) c_neg = rng.normal(0.0, 0.25, size=n) sigma = rng.uniform(0.15, 0.55, size=(n, 1)) innov = rng.normal(0.0, 1.0, size=(n, L)) * sigma return _jit_setar( np.ascontiguousarray(innov), np.ascontiguousarray(phi_pos), np.ascontiguousarray(phi_neg), np.ascontiguousarray(c_pos), np.ascontiguousarray(c_neg), ) def emit_chaotic(rng: np.random.Generator, n: int, L: int) -> np.ndarray: # Milder maps (less fully chaotic) so a tiny residual mass is not pure noise. pick_sine = rng.random(n) < 0.55 r_log = rng.uniform(3.4, 3.9, size=n) r_sin = rng.uniform(0.80, 0.97, size=n) x0 = rng.uniform(0.05, 0.95, size=n) return _jit_chaos( np.ascontiguousarray(pick_sine), np.ascontiguousarray(r_log), np.ascontiguousarray(r_sin), np.ascontiguousarray(x0), L, ) def emit_rff(rng: np.random.Generator, n: int, L: int, *, features: int = 24) -> np.ndarray: """Stationary GP via RFF; longer lengthscales → smoother, more forecastable. K=24 matches the v16fast quality-neutral speed cut — same smoothness class, ~1.5x fewer feature loops so GP mass does not starve the wall. """ t = np.arange(L, dtype=np.float64)[None, :] lengthscale = rng.uniform(40.0, 320.0, size=(n, 1)) acc = np.zeros((n, L), dtype=np.float64) scale = np.sqrt(2.0 / features) for _ in range(features): omega = rng.normal(0.0, 1.0, size=(n, 1)) / lengthscale phase = rng.uniform(0.0, 2.0 * np.pi, size=(n, 1)) np.add(acc, np.cos(omega * t + phase), out=acc) return scale * acc def emit_intermittent(rng: np.random.Generator, n: int, L: int) -> np.ndarray: """Zero-inflated demand with a mild weekly occurrence modulation (retail-like).""" t = np.arange(L, dtype=np.float64)[None, :] p0 = rng.uniform(0.06, 0.35, size=(n, 1)) week = 0.55 + 0.45 * ( 0.5 + 0.5 * np.sin(2.0 * np.pi * t / 7.0 + rng.uniform(0.0, 2.0 * np.pi, size=(n, 1))) ) p = np.clip(p0 * week, 0.02, 0.55) occur = (rng.random((n, L)) < p).astype(np.float64) mag = rng.gamma(shape=2.0, scale=1.0, size=(n, L)) * rng.uniform(1.0, 8.0, size=(n, 1)) floor = rng.uniform(0.0, 0.4, size=(n, 1)) return floor + occur * mag def emit_pulse(rng: np.random.Generator, n: int, L: int) -> np.ndarray: base = emit_rff(rng, n, L, features=24) * rng.uniform(0.5, 2.0, size=(n, 1)) base += _harmonics(rng, n, L, depth=1) * rng.uniform(0.2, 1.0, size=(n, 1)) spikes = _jumps(rng, n, L, rate=4.0 / L, scale=rng.uniform(2.5, 6.5, size=n)) series = base + spikes hold = rng.random((n, L)) < (2.0 / L) hold[:, 0] = False _jit_holds(series, hold) return series # ── weather: sliced archetype mixture (base / radiation / cloud) ──────────── def _cloud_cover(rng: np.random.Generator, n: int, L: int) -> np.ndarray: """U-shaped [0, 100] cloud fraction via multi-timescale red noise + clip.""" t = np.arange(L, dtype=np.float64)[None, :] def _z_ar(phi_lo: float, phi_hi: float) -> np.ndarray: phi = rng.uniform(phi_lo, phi_hi, size=n) z = _ar1(rng.normal(0.0, 1.0, size=(n, L)), phi) return (z - z.mean(axis=1, keepdims=True)) / (z.std(axis=1, keepdims=True) + 1e-9) syn = _z_ar(0.990, 0.9990) mid = _z_ar(0.895, 0.95) fast = _z_ar(0.75, 0.87) g = rng.normal(0.0, 1.0, size=(n, L)) spike = (rng.random((n, L)) < 0.05) * rng.uniform(5.0, 11.0, size=(n, L)) heavy = g * (1.0 + spike) rough = heavy - 0.30 * np.concatenate([np.zeros((n, 1)), heavy[:, :-1]], axis=1) rough = rough / (rough.std(axis=1, keepdims=True) + 1e-9) diur = np.sin(2.0 * np.pi * t / 24.0 + rng.uniform(0.0, 2.0 * np.pi, size=(n, 1))) u = ( 0.95 * rng.uniform(0.8, 1.2, size=(n, 1)) * syn + 0.82 * rng.uniform(0.7, 1.3, size=(n, 1)) * mid + 0.55 * rng.uniform(0.7, 1.3, size=(n, 1)) * fast + 0.30 * rng.uniform(0.7, 1.3, size=(n, 1)) * rough + 0.44 * rng.uniform(0.3, 1.4, size=(n, 1)) * diur ) centre = np.clip(rng.normal(61.0, 29.0, size=(n, 1)), 2.0, 98.0) span = 57.0 * rng.uniform(0.82, 1.18, size=(n, 1)) return np.round(np.clip(centre + span * u, 0.0, 100.0)) def emit_weather(rng: np.random.Generator, n: int, L: int) -> np.ndarray: """Mixture of diurnal base, night-floored radiation, and cloud cover. Archetypes assigned first; expensive multi-scale red-noise only on cloud rows. v2: quieter AR innov + optional monthly harmonic for longer contexts. """ t = np.arange(L, dtype=np.float64)[None, :] is_rad = rng.random(n) < 0.25 is_cloud = (rng.random(n) < 0.24) & ~is_rad is_wbp = ~is_cloud out = np.empty((n, L), dtype=np.float64) wbp = np.nonzero(is_wbp)[0] m = wbp.size if m: rad_rows = is_rad[wbp][:, None] level = rng.normal(0.0, 1.0, size=(m, 1)) a1 = rng.uniform(0.55, 2.1, size=(m, 1)) p1 = rng.uniform(0.0, 2.0 * np.pi, size=(m, 1)) a2 = rng.uniform(0.12, 0.65, size=(m, 1)) p2 = rng.uniform(0.0, 2.0 * np.pi, size=(m, 1)) seas = a1 * np.sin(2.0 * np.pi * t / 24.0 + p1) + a2 * np.sin(2.0 * np.pi * t / 12.0 + p2) weekly_on = (rng.random((m, 1)) < 0.45).astype(np.float64) seas += weekly_on * rng.uniform(0.12, 0.55, size=(m, 1)) * np.sin( 2.0 * np.pi * t / 168.0 + rng.uniform(0.0, 2.0 * np.pi, size=(m, 1)) ) # Monthly harmonic (hourly ~720) — multi-week structure inside L=4096 month_on = (rng.random((m, 1)) < 0.35).astype(np.float64) seas += month_on * rng.uniform(0.08, 0.40, size=(m, 1)) * np.sin( 2.0 * np.pi * t / 720.0 + rng.uniform(0.0, 2.0 * np.pi, size=(m, 1)) ) y_amp = rng.uniform(0.25, 1.6, size=(m, 1)) y_per = rng.uniform(2000.0, 9000.0, size=(m, 1)) yearly = y_amp * np.sin(2.0 * np.pi * t / y_per + rng.uniform(0.0, 2.0 * np.pi, size=(m, 1))) phi = rng.uniform(0.65, 0.96, size=m) sigma = rng.uniform(0.04, 0.18, size=(m, 1)) noise = _ar1(rng.normal(0.0, 1.0, size=(m, L)) * sigma, phi) base = level + seas + yearly + noise thr = rng.uniform(0.2, 0.6, size=(m, 1)) * a1 ramp = rng.uniform(0.85, 1.65, size=(m, 1)) diurnal = ramp * np.maximum(a1 * np.sin(2.0 * np.pi * t / 24.0 + p1) - thr, 0.0) night = diurnal <= 0.0 rad = level + diurnal + yearly + np.where(night, noise * 0.12, noise) out[wbp] = np.where(rad_rows, rad, base) cloud = np.nonzero(is_cloud)[0] if cloud.size: out[cloud] = _cloud_cover(rng, int(cloud.size), L) return out # ── generator entry point ─────────────────────────────────────────────────── class Generator(DataGenerator): """Mixture-of-priors synthesizer. Submit as ``generator.Generator``.""" def __init__(self, config_dir: str, *, seed: int) -> None: cfg_path = Path(config_dir) / "config.json" cfg: dict[str, Any] = ( json.loads(cfg_path.read_text(encoding="utf-8")) if cfg_path.is_file() else {} ) self._cfg = cfg self._seed = int(seed) self._min_len = int(cfg.get("min_length", 64)) self._max_len = int(cfg.get("max_length", 4096)) if self._min_len < 1 or self._max_len < self._min_len: raise ValueError(f"invalid length band [{self._min_len}, {self._max_len}]") mass = dict(_PRIOR_MASS) for key, val in dict(cfg.get("family_weights", {})).items(): if key in mass: mass[key] = float(val) weights = np.asarray([mass[name] for name in _PRIOR_NAMES], dtype=np.float64) if not np.all(np.isfinite(weights)) or weights.min() < 0 or weights.sum() <= 0: raise ValueError("family_weights must be finite, non-negative, and not all zero") self._weights = weights / weights.sum() self._tr_hi = float(cfg.get("tr_hi_frac", 0.20)) self._tr_lo = float(cfg.get("tr_exc_lo", 0.35)) self._tr_hi_s = float(cfg.get("tr_exc_hi", 2.0)) self._gr_lo = float(cfg.get("gr_exc_lo", 0.25)) self._gr_hi = float(cfg.get("gr_exc_hi", 1.4)) self._fixed = self._min_len == self._max_len @property def name(self) -> str: return str(self._cfg.get("name", "aurora-blend-v2")) def _builders(self) -> tuple[Callable[..., np.ndarray], ...]: return ( lambda rng, n, L: emit_trend_seasonal( rng, n, L, hi_frac=self._tr_hi, exc_lo=self._tr_lo, exc_hi=self._tr_hi_s ), emit_regime, lambda rng, n, L: emit_multiplicative( rng, n, L, hi_frac=self._tr_hi, exc_lo=self._gr_lo, exc_hi=self._gr_hi ), emit_ar2, emit_integrated, emit_threshold, emit_chaotic, emit_rff, emit_intermittent, emit_pulse, emit_weather, ) def generate(self, n_series: int) -> Iterator[np.ndarray]: if n_series <= 0: return rng = np.random.default_rng(self._seed) builders = self._builders() L_max = self._max_len done = 0 while done < n_series: # Always draw a full batch so series i is a pure function of (seed, i). lengths = rng.integers(self._min_len, L_max + 1, size=_BATCH) picks = rng.choice(len(_PRIOR_NAMES), size=_BATCH, p=self._weights) slots: list[np.ndarray | None] = [None] * _BATCH for fam_id, build in enumerate(builders): idx = np.nonzero(picks == fam_id)[0] if idx.size == 0: continue block = _finite(build(rng, int(idx.size), L_max)) if self._fixed: for row, slot in enumerate(idx): slots[int(slot)] = np.ascontiguousarray(block[row], dtype=np.float64) else: for row, slot in enumerate(idx): L = int(lengths[slot]) slots[int(slot)] = np.ascontiguousarray(block[row, :L], dtype=np.float64) take = min(_BATCH, n_series - done) for arr in slots[:take]: if arr is None: # pragma: no cover raise RuntimeError("unfilled series slot") yield arr done += take