"""zenfro_v4 — backbone + Grammar of Time + horizon priors. Built on these techniques(coupled calendar seasonality, heavy-tailed / clustered integrated paths, TSMixup, prefix padding, proportional sanitize) with the competitive prior from zenfro_v3: **Grammar of Time (GoT)**. Cascade holds Toto2 fixed and scores the *data prior* on held-out windows: 4096-step context → next 64 steps (CRPS + MASE). Winning synthetic priors (Chronos-2, TempoPFN, CauKer) compose temporal primitives. covers many primitives as mutually exclusive families; GoT allocates mass to **productions** that combine those stems the way real series combine them. Temporal grammar (informal CFG):: Series → Production | Stem Production → Compose | Splice | Nested | Causal | Horizon | Motif Compose → Stem ⊕ Stem [⊕ Stem] # additive / multiplicative phrase Splice → Stem ‖ Stem # clause = change of generating law Nested → Envelope ⋉ Carrier # slow modulates fast Causal → Driver ▷ Response # lagged temporal chain Horizon → Signal + short residual # 64-step forecastable structure Motif → Tile(local shape) # non-sinusoidal repeating phrases Stem → {trend_seasonal, ar2, integrated, spectral, ou, …} Affix → jump | hold | pulse | artifact Determinism, code-only, bounded+finite contracts unchanged: one ``np.random.default_rng(seed)``, allowlisted NumPy/SciPy, ``_sanitize`` gate. """ from __future__ import annotations import json from collections.abc import Iterator from functools import lru_cache, partial from pathlib import Path from queue import Full, Queue from threading import Event, Thread import numpy as np from scipy.signal import lfilter from cascade.interface import DataGenerator # Series generated per vectorised batch. Bounds peak memory to O(_CHUNK · max_len) # so streaming feed modes (which request millions of series and stop early) never # materialise the full corpus. Prefetching holds at most two completed chunks # (current + queued) while the producer may build the next. The base block is # 2048 × 4096 × 8 B = 64 MiB per base family block, plus temporary arrays. # This remains comfortably below the 4 GiB sandbox cap. On the reference local # A100 environment, 2048 rows generated ~6% more points/s than 1024 while 4096 # regressed slightly, so 2048 is the measured throughput sweet spot. _CHUNK = 2560 # Multi-cadence seasonal bank. The lagged 2026-07-21 pool exposed periods # 7/15/24/48/60/96/144/240/288; 15, 60, and 240 were gaps in cascade9. # Longer generic cadences remain for transfer rather than copying one pool. _SEASONAL_PERIODS = np.array( [4, 7, 12, 15, 24, 30, 48, 52, 60, 90, 96, 144, 168, 183, 240, 288, 336, 365, 672, 730], dtype=np.float64, ) _SEASONAL_PROBS = np.array( [0.02, 0.13, 0.03, 0.03, 0.10, 0.03, 0.07, 0.02, 0.06, 0.02, 0.10, 0.06, 0.07, 0.02, 0.06, 0.06, 0.03, 0.04, 0.03, 0.02], dtype=np.float64, ) _SEASONAL_PROBS /= _SEASONAL_PROBS.sum() # Coupled calendar periods teach daily/weekly and short/long cadence # interactions explicitly. Every value is already in _SEASONAL_PERIODS, so the # cached sine/cosine bank remains the only trigonometric work. _SEASONAL_PAIRS = np.array( [[15, 60], [60, 240], [24, 168], [48, 336], [96, 672], [7, 365], [12, 52]], dtype=np.float64, ) # ── family mixture ────────────────────────────────────────────────────────── # Names are the process families the corpus mixes over; the default weights are # a deliberate spread (no single family dominates). Override with # ``"family_weights": {"chaotic": 0.2, ...}`` in config.json to tune the prior # without touching code — unspecified families keep their default weight. _FAMILIES: tuple[str, ...] = ( "trend_seasonal_ar", "regime_shift", "multiplicative", "ar2", "integrated", "threshold_ar", "chaotic", "spectral_gp", "long_memory", "ou_stochastic_vol", "physical_sensors", "seasonal_counts", "intermittent", "pulse_outlier", "got_compose", # Stem ⊕ Stem [⊕ Stem] "got_splice", # Stem ‖ Stem at clause boundaries "got_nested", # Envelope ⋉ Carrier (multi-scale) "got_causal", # Driver ▷ lagged Response "got_horizon", # NEW: forecast-horizon signal + short residual "got_motif", # NEW: tiled local motifs (non-sinusoidal) ) # Dynamics-heavy core retained; ~24% mass on GoT productions so # the model learns combination structure without drowning proven stems. # Horizon/motif explicitly target the 4096→64 eval geometry. _DEFAULT_WEIGHTS: dict[str, float] = { "trend_seasonal_ar": 0.10, "regime_shift": 0.10, "multiplicative": 0.06, "ar2": 0.11, "integrated": 0.09, "threshold_ar": 0.05, "chaotic": 0.02, "spectral_gp": 0.055, "long_memory": 0.045, "ou_stochastic_vol": 0.075, "physical_sensors": 0.01, "seasonal_counts": 0.015, "intermittent": 0.01, "pulse_outlier": 0.01, "got_compose": 0.065, "got_splice": 0.045, "got_nested": 0.04, "got_causal": 0.03, "got_horizon": 0.04, "got_motif": 0.03, } class Generator(DataGenerator): """Grammar-of-Time mixture on the backbone. Submit as ``generator.Generator``.""" def __init__(self, config_dir: str, *, seed: int) -> None: cfg_path = Path(config_dir) / "config.json" cfg = 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)) # = [training] context_length (train on full context) if self._min_len < 1 or self._max_len < self._min_len: raise ValueError(f"invalid length band [{self._min_len}, {self._max_len}]") weights = dict(_DEFAULT_WEIGHTS) for k, v in dict(cfg.get("family_weights", {})).items(): if k in weights: weights[k] = float(v) w = np.asarray([weights[f] for f in _FAMILIES], dtype=np.float64) if not np.all(np.isfinite(w)) or w.min() < 0 or w.sum() <= 0: raise ValueError("family_weights must be finite, non-negative, and not all zero") self._weights = w / w.sum() # v3.9 length-NORMALIZED bimodal trend knobs (trend excursion is length-invariant; # real trend-strength is ~0.02 and length-invariant, but v2's slope*t grows with L). self._tr_hi_frac = float(cfg.get("tr_hi_frac", 0.25)) self._tr_exc_lo = float(cfg.get("tr_exc_lo", 0.4)) self._tr_exc_hi = float(cfg.get("tr_exc_hi", 3.0)) self._gr_exc_lo = float(cfg.get("gr_exc_lo", 0.3)) self._gr_exc_hi = float(cfg.get("gr_exc_hi", 2.0)) self._sa_clean_frac = float(cfg.get("sa_clean_frac", 0.4)) self._sa_clean_lo = float(cfg.get("sa_clean_lo", 0.02)) self._sa_clean_hi = float(cfg.get("sa_clean_hi", 0.12)) self._integrated_heavy_frac = float( cfg.get("integrated_heavy_frac", 0.25) ) self._integrated_sv_frac = float(cfg.get("integrated_sv_frac", 0.30)) self._augment = dict(cfg.get("augment", {})) for name, value in ( ("integrated_heavy_frac", self._integrated_heavy_frac), ("integrated_sv_frac", self._integrated_sv_frac), ("augment.tsmixup", float(self._augment.get("tsmixup", 0.0))), ("augment.pad_prefix", float(self._augment.get("pad_prefix", 0.0))), ): if not 0.0 <= value <= 1.0: raise ValueError(f"{name} must be in [0, 1]") # GoT knobs — composition depth, splice density, nest/horizon/motif. self._got_depth = int(cfg.get("got_depth", 3)) self._got_mul_frac = float(cfg.get("got_mul_frac", 0.35)) self._got_splice_cuts = int(cfg.get("got_splice_cuts", 2)) self._got_nest_ratio = float(cfg.get("got_nest_ratio", 6.0)) self._got_causal_lag_frac = float(cfg.get("got_causal_lag_frac", 0.08)) self._got_horizon_steps = int(cfg.get("got_horizon_steps", 64)) self._got_motif_max = int(cfg.get("got_motif_max", 96)) self._artifact_scale = float(cfg.get("artifact_scale", 1.0)) @property def name(self) -> str: return str(self._cfg.get("name", "zenfro-v4-grammar-of-time")) def generate(self, n_series: int) -> Iterator[np.ndarray]: # Lazy, chunked generation. This is REQUIRED for the streaming feed # modes (chain.toml ``corpus_mode = "stream_cpu"``): the trainer calls # ``generate(n_upper)`` with ``n_upper = token_budget // min_length + 2`` # — often millions — and stops pulling once the token budget is hit # (see cascade/trainer/stream.py). Materialising all ``n_series`` up # front would OOM before the first yield. Generating one CHUNK at a time # keeps memory at O(CHUNK) and stops early when the consumer stops, # while a fixed draw order keeps the whole sequence seed-deterministic. if n_series <= 0: return rng = np.random.default_rng(self._seed) max_len = self._max_len # Bind the trend-excursion knobs as explicit builder arguments (no shared # module state) so the corpus is a pure function of (seed, config). builders = ( partial(_trend_seasonal_ar, hi_frac=self._tr_hi_frac, exc_lo=self._tr_exc_lo, exc_hi=self._tr_exc_hi, clean_frac=self._sa_clean_frac, clean_lo=self._sa_clean_lo, clean_hi=self._sa_clean_hi), _regime_shift, partial(_multiplicative, hi_frac=self._tr_hi_frac, exc_lo=self._gr_exc_lo, exc_hi=self._gr_exc_hi), _ar2, partial( _integrated, heavy_frac=self._integrated_heavy_frac, sv_frac=self._integrated_sv_frac, ), _threshold_ar, _chaotic, _spectral_gp, _long_memory, _ou_stochastic_vol, _physical_sensors, _seasonal_counts, _intermittent, _pulse_outlier, partial(_got_compose, depth=self._got_depth, mul_frac=self._got_mul_frac, hi_frac=self._tr_hi_frac, exc_lo=self._tr_exc_lo, exc_hi=self._tr_exc_hi), partial(_got_splice, n_cuts=self._got_splice_cuts, hi_frac=self._tr_hi_frac, exc_lo=self._tr_exc_lo, exc_hi=self._tr_exc_hi), partial(_got_nested, nest_ratio=self._got_nest_ratio), partial(_got_causal, lag_frac=self._got_causal_lag_frac), partial(_got_horizon, horizon=self._got_horizon_steps, hi_frac=self._tr_hi_frac, exc_lo=self._tr_exc_lo, exc_hi=self._tr_exc_hi, clean_frac=self._sa_clean_frac, clean_lo=self._sa_clean_lo, clean_hi=self._sa_clean_hi), partial(_got_motif, motif_max=self._got_motif_max, hi_frac=self._tr_hi_frac, exc_lo=self._tr_exc_lo, exc_hi=self._tr_exc_hi), ) # Generate one chunk ahead on a CPU thread while the consumer trains on # the current chunk. The isolation benchmark measured 21.9% of training # wall blocked in next(); a one-slot queue overlaps NumPy/SciPy work # (which releases the GIL) without changing the RNG owner or draw order. queue: Queue[object] = Queue(maxsize=1) stop = Event() done = object() def put(item: object) -> bool: while not stop.is_set(): try: queue.put(item, timeout=0.1) return True except Full: continue return False def produce() -> None: try: produced = 0 while produced < n_series and not stop.is_set(): # Always draw a FULL _CHUNK (yielding only what's still # needed), so series i remains a pure function of (seed, i). lengths = rng.integers( self._min_len, max_len + 1, size=_CHUNK ) fam_ids = rng.choice( len(_FAMILIES), size=_CHUNK, p=self._weights ) chunk: list[np.ndarray | None] = [None] * _CHUNK for fam in range(len(_FAMILIES)): idx = np.nonzero(fam_ids == fam)[0] if idx.size == 0: continue block = builders[fam](rng, int(idx.size), max_len) # Preserve positivity for count/magnitude families. preserve_nonnegative = fam in (2, 10, 11, 12) block = _sanitize( _measurement_artifacts( rng, block, preserve_nonnegative=preserve_nonnegative, rate_scale=self._artifact_scale, ) ) for row, series_i in enumerate(idx): length = int(lengths[series_i]) chunk[series_i] = np.ascontiguousarray( block[row, :length], dtype=np.float64 ) # Mix a conservative share of complete, full-context rows # across families. This follows the useful augmentation in # longrange-sv while avoiding it for variable-length # configs, where alignment would be ambiguous. if self._min_len == max_len: mix_rate = float(self._augment.get("tsmixup", 0.0)) mixed = np.nonzero(rng.random(_CHUNK) < mix_rate)[0] for series_i in mixed: source = chunk[series_i] if source is None: # pragma: no cover - defensive continue n_other = int(rng.integers(1, 3)) others = rng.integers(0, _CHUNK, size=n_other) weights = rng.dirichlet(np.ones(n_other + 1)) combined = weights[0] * source valid = True for j, other_i in enumerate(others): other = chunk[int(other_i)] if other is None: # pragma: no cover - defensive valid = False break combined = combined + weights[j + 1] * other if valid: chunk[series_i] = _sanitize(combined) # Constant prefixes represent late-starting sensors and # left-padded histories without changing forecast-tail # dynamics. pad_rate = float(self._augment.get("pad_prefix", 0.0)) padded = np.nonzero(rng.random(_CHUNK) < pad_rate)[0] for series_i in padded: series = chunk[series_i] if series is None or series.size < 8: continue cut = int(rng.integers(series.size // 8, 3 * series.size // 4)) series[:cut] = series[cut] take = min(_CHUNK, n_series - produced) if not put((chunk, take)): return produced += take except BaseException as exc: # propagate producer failures put(exc) finally: put(done) producer = Thread(target=produce, name="zenfro-v4-generator", daemon=True) producer.start() try: while True: item = queue.get() if item is done: break if isinstance(item, BaseException): raise item chunk, take = item for arr in chunk[:take]: # fam_ids partitions [0, _CHUNK); fail loud if that changes. if arr is None: # pragma: no cover - defensive raise RuntimeError("internal: unfilled series slot") yield arr finally: stop.set() producer.join(timeout=1.0) # ── shared vectorised primitives ──────────────────────────────────────────── def _ar1_batch(innov: np.ndarray, phi: np.ndarray) -> np.ndarray: """AR(1) filter applied along the time axis of a (n, L) innovation block. ``x[:, t] = phi * x[:, t-1] + innov[:, t]``. The loop is over time (L iterations, vectorised across the batch), never over the n series. """ n, L = innov.shape x = np.empty((n, L), dtype=np.float64) p = phi.reshape(n) for i in range(n): x[i] = lfilter([1.0], [1.0, -float(p[i])], innov[i]) return x def _ar2_batch(innov: np.ndarray, a1: np.ndarray, a2: np.ndarray) -> np.ndarray: """AR(2) filter: ``x_t = a1 x_{t-1} + a2 x_{t-2} + e_t`` (batched over n).""" n, L = innov.shape x = np.empty((n, L), dtype=np.float64) for i in range(n): x[i] = lfilter( [1.0], [1.0, -float(a1[i]), -float(a2[i])], innov[i] ) return x @lru_cache(maxsize=4) def _seasonal_basis(L: int) -> tuple[np.ndarray, np.ndarray]: """Cached unit sine/cosine waves for the fixed cadence bank.""" angle = ( 2.0 * np.pi * np.arange(L, dtype=np.float64)[None, :] / _SEASONAL_PERIODS[:, None] ) return np.sin(angle), np.cos(angle) def _seasonal(rng: np.random.Generator, n: int, L: int, k_max: int = 3) -> np.ndarray: """Sum of 1..k_max stationary or slowly modulated seasonal components.""" t = np.arange(L, dtype=np.float64)[None, :] sin_basis, cos_basis = _seasonal_basis(L) k = rng.integers(1, k_max + 1, size=n) pair = _SEASONAL_PAIRS[ rng.integers(0, len(_SEASONAL_PAIRS), size=n) ] use_pair = rng.random(n) < 0.35 out = np.zeros((n, L), dtype=np.float64) for j in range(k_max): active = np.nonzero(k > j)[0] per = rng.choice( _SEASONAL_PERIODS, size=n, p=_SEASONAL_PROBS ) if j < 2: per = np.where(use_pair, pair[:, j], per) per = per[:, None] amp = rng.uniform(0.2, 2.0, size=n)[:, None] phase = rng.uniform(0.0, 2.0 * np.pi, size=n)[:, None] # Draw parameters for every row to preserve the fixed RNG sequence, but # evaluate only active rows. Stationary components reuse the cadence # bank via sin(a+b), avoiding a fresh transcendental pass over n×L. basis_idx = np.searchsorted(_SEASONAL_PERIODS, per[active, 0]) component = amp[active] * ( sin_basis[basis_idx] * np.cos(phase[active]) + cos_basis[basis_idx] * np.sin(phase[active]) ) # Real seasonal strength and timing drift. TempoPFN's strongest # non-SDE ablation was its complex-seasonality prior, so a minority of # components receive slow amplitude and phase modulation while the # stationary baseline remains well represented. modulated = np.nonzero((k > j) & (rng.random(n) < 0.35))[0] if modulated.size: # Map global row indices into the active component block. modulated_local = np.searchsorted(active, modulated) modulated_arg = ( 2.0 * np.pi * t / per[modulated] + phase[modulated] ) m_per = np.clip( per[modulated] * rng.uniform( 4.0, 12.0, size=(modulated.size, 1) ), 32.0, 2.0 * L, ) m_phase = rng.uniform( 0.0, 2.0 * np.pi, size=(modulated.size, 1) ) slow = np.sin(2.0 * np.pi * t / m_per + m_phase) amp_mod = 1.0 + rng.uniform( 0.05, 0.45, size=(modulated.size, 1) ) * slow phase_mod = rng.uniform( 0.05, 0.75, size=(modulated.size, 1) ) * np.sin(2.0 * np.pi * t / (1.7 * m_per) - m_phase) component[modulated_local] = ( amp[modulated] * amp_mod * np.sin(modulated_arg + phase_mod) ) out[active] += component return out def _sparse_jumps(rng: np.random.Generator, n: int, L: int, rate: float, scale) -> np.ndarray: """A (n, L) block of mostly-zero values with occasional N(0, scale) jumps. ``cumsum`` over this yields a piecewise-constant level; ``exp(cumsum)`` of a scaled version yields a piecewise-constant positive multiplier. """ mask = rng.random((n, L)) < rate mask[:, 0] = False rows, cols = np.nonzero(mask) jumps = np.zeros((n, L), dtype=np.float64) if rows.size == 0: return jumps # Rates are O(1/L), so draw magnitudes only for actual events rather than # allocating and filling a second dense n×L normal array. s = np.asarray(scale, dtype=np.float64) event_scale = s if s.ndim == 0 else s.reshape(n)[rows] jumps[rows, cols] = rng.normal(0.0, 1.0, size=rows.size) * event_scale return jumps def _row_standardize(x: np.ndarray) -> np.ndarray: x = x - x.mean(axis=1, keepdims=True) sd = x.std(axis=1, keepdims=True) return x / np.where(sd < 1e-12, 1.0, sd) def _measurement_artifacts( rng: np.random.Generator, block: np.ndarray, *, preserve_nonnegative: bool, rate_scale: float = 1.0, ) -> np.ndarray: """Apply sparse, cheap real-measurement effects to a generated block. TempoPFN reports a 5.4% aggregate CRPS gain from its complete augmentation pipeline, but does not isolate optimal probabilities for Toto2. These rates are deliberately conservative: most rows remain untouched, and a selected row receives only plausible reversal/sign, censoring, quantization, or sample-and-hold behavior. ``rate_scale`` multiplies base rates. """ original = np.asarray(block, dtype=np.float64) out = original.copy() n, L = out.shape rs = float(np.clip(rate_scale, 0.0, 3.0)) reverse = rng.random(n) < (0.06 * rs) out[reverse] = out[reverse, ::-1] if not preserve_nonnegative: invert = rng.random(n) < (0.04 * rs) out[invert] *= -1.0 # Sensor saturation / floor effects. Existing sample values are used as # thresholds, avoiding artificial scales and preserving integer counts. for row in np.nonzero(rng.random(n) < (0.06 * rs))[0]: q = float(rng.uniform(0.03, 0.18)) if rng.random() < 0.5: out[row] = np.minimum(out[row], np.quantile(out[row], 1.0 - q)) else: out[row] = np.maximum(out[row], np.quantile(out[row], q)) quantized = np.nonzero(rng.random(n) < (0.07 * rs))[0] if quantized.size: x = out[quantized] lo = x.min(axis=1, keepdims=True) hi = x.max(axis=1, keepdims=True) levels = rng.integers(16, 257, size=(quantized.size, 1)) step = (hi - lo) / np.maximum(levels - 1, 1) safe_step = np.where(step < 1e-12, 1.0, step) out[quantized] = lo + np.rint((x - lo) / safe_step) * safe_step # Zero-order-hold resampling approximates telemetry gathered at a lower # cadence and forwarded at the nominal cadence. held = np.nonzero(rng.random(n) < (0.04 * rs))[0] if held.size: factors = rng.choice([2, 4, 8], size=held.size, p=[0.55, 0.30, 0.15]) for factor in (2, 4, 8): rows = held[factors == factor] if rows.size: out[rows] = np.repeat( out[rows, ::factor], factor, axis=1 )[:, :L] # Heavy zero inflation plus upper censoring can otherwise collapse a sparse # row to its baseline. Such a row carries no forecasting signal. degenerate = out.std(axis=1) < 1e-9 out[degenerate] = original[degenerate] return out # ── family builders: each returns a (n, L) float64 block ──────────────────── def _trend_seasonal_ar(rng: np.random.Generator, n: int, L: int, *, hi_frac: float = 0.25, exc_lo: float = 0.4, exc_hi: float = 3.0, clean_frac: float = 0.4, clean_lo: float = 0.02, clean_hi: float = 0.12) -> np.ndarray: t = np.arange(L, dtype=np.float64)[None, :] level = rng.normal(0.0, 1.0, size=(n, 1)) # v3: bimodal trend. The total trend EXCURSION over the series is drawn directly # (0..exc across t/(L-1)), so the trend sits ~16x below v2's slope*t — v2's linear # trend was a measured ~16x too strong vs real data at production lengths. _hi = rng.random((n, 1)) < hi_frac exc = np.where(_hi, rng.normal(0.0, exc_hi, size=(n, 1)), rng.normal(0.0, exc_lo, size=(n, 1))) tn = t / max(L - 1, 1) series = level + exc * tn + _seasonal(rng, n, L) phi = rng.uniform(0.0, 0.85, size=n) clean = rng.random((n, 1)) < clean_frac sigma = np.where( clean, rng.uniform(clean_lo, clean_hi, size=(n, 1)), rng.uniform(0.1, 0.6, size=(n, 1)), ) innov = rng.normal(0.0, 1.0, size=(n, L)) * sigma return series + _ar1_batch(innov, phi) def _regime_shift(rng: np.random.Generator, n: int, L: int) -> np.ndarray: # Piecewise-constant level via cumsum of sparse jumps, plus a piecewise # variance regime (occasional volatility multiplier), plus mild seasonality. level = np.cumsum(_sparse_jumps(rng, n, L, rate=3.0 / L, scale=2.0), axis=1) log_vol = np.cumsum(_sparse_jumps(rng, n, L, rate=3.0 / L, scale=0.5), axis=1) vol = np.exp(np.clip(log_vol, -3.0, 3.0)) * rng.uniform(0.1, 0.5, size=(n, 1)) noise = rng.normal(0.0, 1.0, size=(n, L)) * vol seas = _seasonal(rng, n, L, k_max=2) * rng.uniform(0.0, 1.0, size=(n, 1)) # Piecewise-affine drift complements abrupt level jumps. Sparse slope # changes create ramps and recoveries without the explosive scale of an I(2) # process, covering TempoPFN's high-impact Step/Sawtooth structures. slope = rng.normal(0.0, 1.0 / L, size=(n, 1)) + np.cumsum( _sparse_jumps(rng, n, L, rate=2.0 / L, scale=4.0 / L), axis=1 ) piecewise_trend = np.cumsum(slope, axis=1) return level + piecewise_trend + seas + noise def _multiplicative(rng: np.random.Generator, n: int, L: int, *, hi_frac: float = 0.25, exc_lo: float = 0.3, exc_hi: float = 2.0) -> np.ndarray: t = np.arange(L, dtype=np.float64)[None, :] # v3: bimodal log-growth excursion (drawn directly), same rationale as the linear trend. _hg = rng.random((n, 1)) < hi_frac gexc = np.where(_hg, 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_level = np.exp(gexc * tn + rng.normal(0.0, 0.3, size=(n, 1))) # positive, drifting amp = rng.uniform(0.1, 0.6, size=(n, 1)) seasonal_shape = _seasonal(rng, n, L, k_max=1) seasonal_sd = seasonal_shape.std(axis=1, keepdims=True) seasonal_shape /= np.where(seasonal_sd < 1e-12, 1.0, seasonal_sd) seas = 1.0 + amp * seasonal_shape noise = 1.0 + rng.normal(0.0, 1.0, size=(n, L)) * rng.uniform(0.02, 0.15, size=(n, 1)) scale = rng.uniform(1.0, 50.0, size=(n, 1)) return scale * base_level * np.clip(seas, 0.05, None) * np.clip(noise, 0.05, None) def _ar2(rng: np.random.Generator, n: int, L: int) -> np.ndarray: # Draw partial autocorrelations in (-1, 1) and map to AR(2) coeffs via # Levinson-Durbin, which guarantees stationarity. Bias p1 high for # persistent (sometimes near-unit-root) series. p1 = rng.uniform(0.3, 0.98, size=n) p2 = rng.uniform(-0.6, 0.6, size=n) a2 = p2 a1 = p1 * (1.0 - p2) sigma = rng.uniform(0.2, 0.8, size=(n, 1)) innov = rng.normal(0.0, 1.0, size=(n, L)) * sigma x = _ar2_batch(innov, a1, a2) drift = rng.normal(0.0, 0.005, size=(n, 1)) * np.arange(L, dtype=np.float64)[None, :] return x + drift def _integrated( rng: np.random.Generator, n: int, L: int, *, heavy_frac: float = 0.25, sv_frac: float = 0.30, ) -> np.ndarray: """I(1)/I(2) paths with selective heavy tails and clustered volatility. The Gaussian baseline remains the majority. Heavy rows use variance-scaled Student-t innovations, while stochastic-volatility rows receive a smooth AR(1) log-vol multiplier. These mechanisms are applied inside an existing cascade9 family rather than funding a new family at the expense of its measured mixture. """ order2 = rng.random(n) < 0.35 drift = rng.normal(0.0, 0.02, size=(n, 1)) sigma = rng.uniform(0.2, 1.0, size=(n, 1)) eps = rng.normal(0.0, 1.0, size=(n, L)) heavy = np.nonzero(rng.random(n) < heavy_frac)[0] if heavy.size: df = rng.uniform(3.0, 12.0, size=(heavy.size, 1)) eps[heavy] = rng.standard_t(df, size=(heavy.size, L)) / np.sqrt( df / (df - 2.0) ) stochastic = np.nonzero(rng.random(n) < sv_frac)[0] if stochastic.size: phi = 0.995 log_vol = lfilter( [1.0], [1.0, -phi], rng.standard_normal((stochastic.size, L)), axis=1, ) log_vol -= log_vol.mean(axis=1, keepdims=True) log_vol /= np.maximum(log_vol.std(axis=1, keepdims=True), 1e-9) log_vol *= rng.uniform(0.10, 0.55, size=(stochastic.size, 1)) eps[stochastic] *= np.exp(np.clip(log_vol, -2.0, 2.0)) steps = eps * sigma + drift walk = np.cumsum(steps, axis=1) walk2 = np.cumsum(walk, axis=1) o2 = order2[:, None] # I(2) grows fast; damp it so it shares scale with the I(1) branch. return np.where(o2, walk2 / max(L, 1) ** 0.5, walk) def _threshold_ar(rng: np.random.Generator, n: int, L: int) -> np.ndarray: # SETAR(2): coefficient flips with the sign of the previous value — a simple # nonlinear recurrence that produces asymmetric, regime-switching dynamics. phi_hi = rng.uniform(0.3, 0.9, size=n) phi_lo = rng.uniform(-0.9, 0.3, size=n) const_hi = rng.normal(0.0, 0.3, size=n) const_lo = rng.normal(0.0, 0.3, size=n) sigma = rng.uniform(0.2, 0.7, size=(n, 1)) innov = rng.normal(0.0, 1.0, size=(n, L)) * sigma x = np.empty((n, L), dtype=np.float64) x[:, 0] = innov[:, 0] for t in range(1, L): prev = x[:, t - 1] hi = prev >= 0.0 phi = np.where(hi, phi_hi, phi_lo) const = np.where(hi, const_hi, const_lo) x[:, t] = np.clip(const + phi * prev + innov[:, t], -1e6, 1e6) return x def _chaotic(rng: np.random.Generator, n: int, L: int) -> np.ndarray: # Bounded chaotic maps: logistic x_{t+1}=r x(1-x) with r∈[3.6,4.0], and the # sine map r sin(pi x). Both stay in [0,1]; standardise afterwards. A random # observation length as a "sampling rate" adds variety across series. use_sine = rng.random(n) < 0.5 r_log = rng.uniform(3.6, 4.0, size=n) r_sin = rng.uniform(0.85, 1.0, size=n) x0 = rng.uniform(0.05, 0.95, size=n) x = np.empty((n, L), dtype=np.float64) cur = x0.copy() x[:, 0] = cur for t in range(1, L): nxt_log = r_log * cur * (1.0 - cur) nxt_sin = r_sin * np.sin(np.pi * cur) cur = np.where(use_sine, nxt_sin, nxt_log) cur = np.clip(cur, 0.0, 1.0) x[:, t] = cur return x def _spectral_gp(rng: np.random.Generator, n: int, L: int) -> np.ndarray: """Smooth stationary GP-like paths sampled in O(n L log L). An RBF kernel has a Gaussian spectral density. Drawing complex Fourier coefficients under that envelope and applying one batched inverse FFT preserves the useful smoothness/length-scale prior without the old 48-pass cosine loop. """ f = np.fft.rfftfreq(L)[None, :] lengthscale = np.exp(rng.uniform(np.log(8.0), np.log(256.0), size=(n, 1))) envelope = np.exp(-0.5 * (2.0 * np.pi * lengthscale * f) ** 2) z = rng.standard_normal((n, f.shape[1])) + 1j * rng.standard_normal((n, f.shape[1])) z[:, 0] = 0.0 x = np.fft.irfft(z * np.sqrt(envelope), n=L, axis=1) sd = x.std(axis=1, keepdims=True) return x / np.where(sd < 1e-12, 1.0, sd) def _long_memory(rng: np.random.Generator, n: int, L: int) -> np.ndarray: """Fractional power-law paths with both persistent and rough regimes. The spectral slope beta spans anti-persistent noise through persistent long-memory levels. A minority of rows are integrated once to include nonstationary fBm-like paths; row standardisation keeps scales bounded. """ f = np.fft.rfftfreq(L) safe_f = np.maximum(f, 1.0 / L)[None, :] beta = rng.uniform(-0.6, 2.4, size=(n, 1)) amp = safe_f ** (-0.5 * beta) # Some rows change roughness above a random frequency, giving smooth # large-scale structure and rough local variation (or the reverse) without # another FFT. Match amplitudes at the split to avoid a spectral jump. multiscale = rng.random((n, 1)) < 0.4 split_idx = rng.integers(8, max(9, f.size // 3), size=(n, 1)) split_f = np.maximum(split_idx / L, 1.0 / L) beta_hi = rng.uniform(-0.6, 2.8, size=(n, 1)) above = np.arange(f.size)[None, :] > split_idx amp_hi = split_f ** (-0.5 * beta) \ * (safe_f / split_f) ** (-0.5 * beta_hi) amp = np.where(multiscale & above, amp_hi, amp) amp[:, 0] = 0.0 z = rng.standard_normal((n, f.size)) + 1j * rng.standard_normal((n, f.size)) x = np.fft.irfft(z * amp, n=L, axis=1) integrate = rng.random(n) < 0.25 if integrate.any(): x[integrate] = np.cumsum(x[integrate], axis=1) x -= x.mean(axis=1, keepdims=True) sd = x.std(axis=1, keepdims=True) return x / np.where(sd < 1e-12, 1.0, sd) def _ou_stochastic_vol(rng: np.random.Generator, n: int, L: int) -> np.ndarray: """Regime-switching mean reversion with bounded stochastic volatility. This is a CPU-cheap discrete Euler/AR analogue of TempoPFN's highest-impact OU SDE prior. Regime paths, seasonal means, volatility envelopes, and heavy-tail masks are sampled in whole blocks; only the state recurrence scans time, vectorised across all rows. """ # Toggle between a fast/quiet and a slow/volatile regime. A cumulative XOR # builds persistent Markov-like paths without a per-row Python loop. switch_rate = np.exp(rng.uniform(np.log(0.001), np.log(0.15), size=(n, 1))) switches = rng.random((n, L)) < switch_rate switches[:, 0] = rng.random(n) < 0.5 regime = np.bitwise_and(np.cumsum(switches, axis=1), 1).astype(np.int8) # One mean-reversion speed per row lets SciPy execute the recurrence in # compiled code. Regime paths still switch equilibrium mean and volatility; # rows span both fast/quiet and slow/persistent reversion rates. slow = rng.random((n, 1)) < 0.5 phi = np.where( slow, rng.uniform(0.995, 0.9995, size=(n, 1)), rng.uniform(0.90, 0.99, size=(n, 1)), ) mu0 = rng.normal(-2.0, 1.0, size=(n, 1)) mu1 = rng.normal(2.0, 1.0, size=(n, 1)) mean = np.where(regime == 0, mu0, mu1) seasonal_on = rng.random((n, 1)) < 0.6 mean += seasonal_on * _seasonal(rng, n, L, k_max=3) \ * rng.uniform(0.5, 3.0, size=(n, 1)) sigma0 = rng.lognormal(np.log(0.3), 0.3, size=(n, 1)) sigma1 = rng.lognormal(np.log(1.5), 0.5, size=(n, 1)) base_sigma = np.where(regime == 0, sigma0, sigma1) log_vol = np.cumsum( _sparse_jumps(rng, n, L, rate=8.0 / L, scale=0.35), axis=1 ) log_vol -= log_vol.mean(axis=1, keepdims=True) vol = base_sigma * np.exp(np.clip(log_vol, -1.5, 1.5)) eps = rng.standard_normal((n, L)) heavy = np.nonzero(rng.random(n) < 0.35)[0] if heavy.size: # Replace only heavy-tailed rows; drawing Student-t noise for every row # previously discarded 65% of that relatively expensive work. eps[heavy] = ( rng.standard_t(4.0, size=(heavy.size, L)) / np.sqrt(2.0) ) shocks = rng.random((n, L)) < (3.0 / L) shock_rows, shock_cols = np.nonzero(shocks) # As with sparse jumps, draw shock magnitudes only at the O(n) events. eps[shock_rows, shock_cols] += rng.normal( 0.0, 5.0, size=shock_rows.size ) innovation_scale = np.sqrt(np.maximum(1.0 - phi * phi, 1e-6)) drive = (1.0 - phi) * mean + innovation_scale * vol * eps out = np.empty((n, L), dtype=np.float64) out[:, 0] = mean[:, 0] + vol[:, 0] * eps[:, 0] for i in range(n): p = float(phi[i, 0]) out[i, 1:] = lfilter( [1.0], [1.0, -p], drive[i, 1:], zi=[p * out[i, 0]] )[0] scale = np.exp(rng.uniform(np.log(0.1), np.log(50.0), size=(n, 1))) shift = rng.uniform(-100.0, 100.0, size=(n, 1)) return out * scale + shift def _physical_sensors(rng: np.random.Generator, n: int, L: int) -> np.ndarray: """Generic physical measurements without matching one private dataset. Four row-level archetypes cover smooth signed measurements, bounded percentages, pressure-like wandering levels, and non-negative skewed magnitudes. All share multi-cadence seasonality, smooth synoptic variation, and sparse fronts/gusts. """ seasonal = _seasonal(rng, n, L, k_max=2) smooth = _spectral_gp(rng, n, L) fronts = np.cumsum( _sparse_jumps(rng, n, L, rate=5.0 / L, scale=1.0), axis=1 ) base = ( seasonal * rng.uniform(0.3, 2.0, size=(n, 1)) + smooth * rng.uniform(0.2, 1.2, size=(n, 1)) + fronts * rng.uniform(0.2, 1.0, size=(n, 1)) ) kind = rng.integers(0, 4, size=n) out = base.copy() bounded = kind == 1 if bounded.any(): gain = rng.uniform(0.8, 3.5, size=(int(bounded.sum()), 1)) midpoint = rng.uniform(-0.8, 0.8, size=(int(bounded.sum()), 1)) out[bounded] = 100.0 / (1.0 + np.exp(-gain * (base[bounded] - midpoint))) pressure = kind == 2 if pressure.any(): count = int(pressure.sum()) walk = np.cumsum(rng.standard_normal((count, L)), axis=1) / np.sqrt(L) level = rng.uniform(900.0, 1100.0, size=(count, 1)) out[pressure] = level + rng.uniform(2.0, 15.0, size=(count, 1)) * walk \ + 2.0 * fronts[pressure] + 0.5 * seasonal[pressure] magnitude = kind == 3 if magnitude.any(): count = int(magnitude.sum()) gusts = (rng.random((count, L)) < (8.0 / L)) \ * rng.lognormal(0.0, 0.8, size=(count, L)) power = rng.uniform(1.0, 1.6, size=(count, 1)) out[magnitude] = np.abs(base[magnitude]) ** power + gusts return out def _seasonal_counts(rng: np.random.Generator, n: int, L: int) -> np.ndarray: """Seasonal Poisson/negative-binomial counts with decaying bursts. This keeps count positivity and discreteness intact while covering overdispersion, cadence-linked rate variation, slow signed growth, and release/news-like bursts. Computation remains batched across rows. """ t = np.arange(L, dtype=np.float64)[None, :] period = rng.choice( _SEASONAL_PERIODS, size=(n, 1), p=_SEASONAL_PROBS ) phase = rng.uniform(0.0, 2.0 * np.pi, size=(n, 1)) amp = rng.uniform(0.15, 0.8, size=(n, 1)) log_rate = amp * np.sin(2.0 * np.pi * t / period + phase) second = rng.random((n, 1)) < 0.55 log_rate += second * (0.5 * amp) * np.sin( 4.0 * np.pi * t / period + rng.uniform(0.0, 2.0 * np.pi, size=(n, 1)) ) # A minority carry explicit calendar interaction: intraday cadence plus # seven day-specific factors, with a randomized weekend dip or lift. calendar = rng.random((n, 1)) < 0.35 day_period = rng.choice([24, 48, 96, 144], size=(n, 1)) day_idx = (np.floor_divide(np.arange(L)[None, :], day_period) % 7).astype(np.int64) day_factors = rng.normal(0.0, 0.12, size=(n, 7)) day_factors[:, 5:] += rng.uniform(-0.8, 0.3, size=(n, 1)) calendar_effect = np.take_along_axis(day_factors, day_idx, axis=1) log_rate += calendar * calendar_effect excursion = rng.uniform(-0.5, 0.5, size=(n, 1)) log_rate += excursion * t / max(L - 1, 1) # Sparse positive impulses filtered by row-specific decay create bursts # without a Python loop over timesteps. impulses = ( (rng.random((n, L)) < (2.0 / L)) * rng.uniform(1.0, 10.0, size=(n, L)) ) burst = _ar1_batch(impulses, rng.uniform(0.85, 0.995, size=(n, 1))) base = np.exp(rng.uniform(np.log(3.0), np.log(3000.0), size=(n, 1))) lam = base * np.exp(np.clip(log_rate, -5.0, 5.0)) * (1.0 + burst) np.clip(lam, 0.0, 1.0e7, out=lam) # A gamma-mixed Poisson is negative-binomial marginally and provides # realistic overdispersion. Half the rows remain ordinary Poisson. overdispersed = rng.random((n, 1)) < 0.5 shape = rng.uniform(0.5, 4.0, size=(n, 1)) mixed = lam * rng.gamma(shape, 1.0 / shape, size=(n, L)) return rng.poisson(np.where(overdispersed, mixed, lam)).astype(np.float64) def _intermittent(rng: np.random.Generator, n: int, L: int) -> np.ndarray: # Seasonal zero-inflated demand. Occurrence probabilities vary by cadence # instead of being iid, teaching the model forecastable sparse structure. t = np.arange(L, dtype=np.float64)[None, :] base_p = rng.uniform(0.03, 0.35, size=(n, 1)) period = rng.choice([7.0, 12.0, 24.0, 48.0, 168.0], size=(n, 1)) season = rng.uniform(0.2, 1.2, size=(n, 1)) * np.sin( 2.0 * np.pi * t / period + rng.uniform(0.0, 2.0 * np.pi, size=(n, 1)) ) logit = np.log(base_p / (1.0 - base_p)) + season p = 1.0 / (1.0 + np.exp(-logit)) occur = (rng.random((n, L)) < p).astype(np.float64) magnitude = ( rng.gamma(shape=2.0, scale=1.0, size=(n, L)) * rng.uniform(1.0, 10.0, size=(n, 1)) * np.exp(0.25 * season) ) baseline = rng.uniform(0.0, 0.5, size=(n, 1)) return baseline + occur * magnitude def _pulse_outlier(rng: np.random.Generator, n: int, L: int) -> np.ndarray: # A smooth base with isolated outliers, persistent shock/recovery responses, # and genuine held-constant runs. base = _spectral_gp(rng, n, L) * rng.uniform(0.5, 2.0, size=(n, 1)) base += _seasonal(rng, n, L, k_max=1) * rng.uniform(0.0, 1.0, size=(n, 1)) sharp = _sparse_jumps( rng, n, L, rate=3.0 / L, scale=rng.uniform(3.0, 8.0, size=n) ) impulses = _sparse_jumps( rng, n, L, rate=2.0 / L, scale=rng.uniform(2.0, 7.0, size=n) ) recovery = _ar1_batch(impulses, rng.uniform(0.75, 0.995, size=n)) series = base + sharp + recovery # Sparse event loops, not a time-axis scan: typically two starts per row. starts = rng.random((n, L)) < (2.0 / L) starts[:, 0] = False for row in range(n): for start in np.nonzero(starts[row])[0]: run = int(rng.integers(3, 65)) end = min(int(start) + run, L) series[row, start:end] = series[row, start - 1] return series # ── Grammar-of-Time productions ───────────────────────────────────────────── # Stem pool for compositions. Order is fixed so RNG draw sequences stay stable # across config-only weight changes to non-GoT families. # Lightweight stem pool for GoT productions. Full cascade9 builders stay as # top-level families; compositions need many stems per row, so these stay # FFT/AR/seasonal only — no Python-over-t recurrences. def _stem_ar_seasonal(rng: np.random.Generator, n: int, L: int) -> np.ndarray: seas = _seasonal(rng, n, L, k_max=2) phi = rng.uniform(0.1, 0.9, size=n) sigma = rng.uniform(0.15, 0.7, size=(n, 1)) innov = rng.normal(0.0, 1.0, size=(n, L)) * sigma return _row_standardize(seas * rng.uniform(0.3, 1.2, size=(n, 1)) + _ar1_batch(innov, phi)) def _stem_integrated(rng: np.random.Generator, n: int, L: int) -> np.ndarray: sigma = rng.uniform(0.2, 1.0, size=(n, 1)) walk = np.cumsum(rng.normal(0.0, 1.0, size=(n, L)) * sigma, axis=1) return _row_standardize(walk) def _sample_stem(rng: np.random.Generator, n: int, L: int) -> np.ndarray: """Draw one cheap stem family per row and fill a (n, L) block.""" builders = ( _stem_ar_seasonal, _spectral_gp, _long_memory, _stem_integrated, _ar2, ) fam = rng.integers(0, len(builders), size=n) out = np.empty((n, L), dtype=np.float64) for k, builder in enumerate(builders): idx = np.nonzero(fam == k)[0] if idx.size == 0: continue out[idx] = _row_standardize(builder(rng, int(idx.size), L)) return out def _got_compose(rng: np.random.Generator, n: int, L: int, *, depth: int = 3, mul_frac: float = 0.35, hi_frac: float = 0.25, exc_lo: float = 0.4, exc_hi: float = 2.5) -> np.ndarray: """Production: Stem ⊕ Stem [⊕ Stem] — additive or multiplicative phrase. Each row stacks ``depth`` standardised stems. A minority use multiplicative agreement (level × seasonal-like factor), matching TempoPFN-style compound structure rather than a single process family. """ depth = int(np.clip(depth, 2, 4)) # Always draw ``depth`` stems so the RNG stream is depth-stable. stems = [_sample_stem(rng, n, L) for _ in range(depth)] n_active = rng.integers(2, depth + 1, size=n) out = np.zeros((n, L), dtype=np.float64) use_mul = rng.random(n) < mul_frac for j, stem in enumerate(stems): active = (n_active > j)[:, None] w = rng.uniform(0.4, 1.6, size=(n, 1)) # Additive branch. add_mask = active & (~use_mul[:, None]) out = np.where(add_mask, out + w * stem, out) # Multiplicative branch: first stem is the carrier; later stems modulate. if j == 0: out = np.where(use_mul[:, None], stem, out) else: factor = 1.0 + 0.35 * w * stem out = np.where(active & use_mul[:, None], out * factor, out) # Mild length-normalised trend affix so compose rows still carry forecastable # drift without exploding scale. t = np.arange(L, dtype=np.float64)[None, :] / max(L - 1, 1) _hi = rng.random((n, 1)) < hi_frac exc = np.where(_hi, rng.normal(0.0, exc_hi * 0.5, size=(n, 1)), rng.normal(0.0, exc_lo * 0.5, size=(n, 1))) out = out + exc * t # Sparse punctuation affix (jumps) on a minority of rows. punct = rng.random(n) < 0.4 if punct.any(): jumps = np.cumsum( _sparse_jumps(rng, n, L, rate=2.5 / L, scale=1.5), axis=1 ) out[punct] = out[punct] + jumps[punct] scale = np.exp(rng.uniform(np.log(0.2), np.log(40.0), size=(n, 1))) shift = rng.uniform(-50.0, 50.0, size=(n, 1)) return out * scale + shift def _got_splice(rng: np.random.Generator, n: int, L: int, *, n_cuts: int = 2, hi_frac: float = 0.25, exc_lo: float = 0.4, exc_hi: float = 2.5) -> np.ndarray: """Production: Stem ‖ Stem — clause boundaries splice different dynamics. Unlike cascade9's piecewise level jumps inside one process, each clause is an independent stem; breakpoints teach structural change of *generating law*. """ n_cuts = int(np.clip(n_cuts, 1, 4)) n_clauses = n_cuts + 1 clauses = [_sample_stem(rng, n, L) for _ in range(n_clauses)] # Cut positions in (0.15L, 0.85L), sorted per row. cuts = np.sort( rng.integers(max(1, L // 8), max(2, (7 * L) // 8), size=(n, n_cuts)), axis=1, ) # Enforce strictly increasing cuts with a small gap. for c in range(1, n_cuts): cuts[:, c] = np.maximum(cuts[:, c], cuts[:, c - 1] + max(8, L // 32)) cuts = np.clip(cuts, 1, L - 2) out = clauses[0].copy() t_idx = np.arange(L)[None, :] for c in range(n_cuts): after = t_idx >= cuts[:, c:c + 1] out = np.where(after, clauses[c + 1], out) # Soft blend near each cut so the splice is a transition, not a hard glitch # (real regime changes often ramp over a few steps). blend_w = max(4, L // 128) for c in range(n_cuts): cut = cuts[:, c:c + 1] dist = (t_idx - cut).astype(np.float64) gate = np.clip(0.5 + dist / (2.0 * blend_w), 0.0, 1.0) left = clauses[c] right = clauses[c + 1] near = np.abs(dist) <= blend_w blended = (1.0 - gate) * left + gate * right out = np.where(near, blended, out) # Optional level offset between clauses (structural break magnitude). level_jump = rng.normal(0.0, 1.5, size=(n, n_cuts)) for c in range(n_cuts): after = t_idx >= cuts[:, c:c + 1] out = np.where(after, out + level_jump[:, c:c + 1], out) t = np.arange(L, dtype=np.float64)[None, :] / max(L - 1, 1) _hi = rng.random((n, 1)) < hi_frac exc = np.where(_hi, rng.normal(0.0, exc_hi * 0.4, size=(n, 1)), rng.normal(0.0, exc_lo * 0.4, size=(n, 1))) out = out + exc * t scale = np.exp(rng.uniform(np.log(0.2), np.log(40.0), size=(n, 1))) shift = rng.uniform(-50.0, 50.0, size=(n, 1)) return out * scale + shift def _got_nested(rng: np.random.Generator, n: int, L: int, *, nest_ratio: float = 6.0) -> np.ndarray: """Production: Envelope ⋉ Carrier — slow scale nests a fast carrier. Hierarchical seasonality / synoptic weather / business-cycle nesting: a smooth long-scale envelope modulates amplitude (and sometimes phase) of a faster seasonal or AR carrier. This is the multi-scale grammar Chronos-style priors emphasise but cascade9 only touches via modulated seasonality. """ nest_ratio = float(np.clip(nest_ratio, 2.0, 24.0)) t = np.arange(L, dtype=np.float64)[None, :] # Slow envelope: spectral GP with long lengthscale, or long sinusoid. use_gp_env = rng.random(n) < 0.55 env = np.empty((n, L), dtype=np.float64) gp_rows = np.nonzero(use_gp_env)[0] sin_rows = np.nonzero(~use_gp_env)[0] if gp_rows.size: # Force long lengthscales for the envelope. f = np.fft.rfftfreq(L)[None, :] lengthscale = np.exp( rng.uniform(np.log(64.0), np.log(min(512.0, L / 2.0)), size=(gp_rows.size, 1)) ) envelope = np.exp(-0.5 * (2.0 * np.pi * lengthscale * f) ** 2) z = rng.standard_normal((gp_rows.size, f.shape[1])) + 1j * rng.standard_normal( (gp_rows.size, f.shape[1]) ) z[:, 0] = 0.0 g = np.fft.irfft(z * np.sqrt(envelope), n=L, axis=1) env[gp_rows] = _row_standardize(g) if sin_rows.size: per = rng.uniform(L / nest_ratio, L / 1.5, size=(sin_rows.size, 1)) phase = rng.uniform(0.0, 2.0 * np.pi, size=(sin_rows.size, 1)) env[sin_rows] = np.sin(2.0 * np.pi * t / per + phase) # Fast carrier: seasonal bank and/or AR(2). carrier = _seasonal(rng, n, L, k_max=3) carrier = _row_standardize(carrier) mix_ar = rng.random(n) < 0.45 if mix_ar.any(): ar = _row_standardize(_ar2(rng, n, L)) w = rng.uniform(0.3, 0.7, size=(n, 1)) carrier = np.where(mix_ar[:, None], w * carrier + (1.0 - w) * ar, carrier) amp = 1.0 + rng.uniform(0.3, 1.4, size=(n, 1)) * env # Phase wobble as a small quadrature mix with a lagged carrier — vectorised, # no per-row roll. Equivalent spirit: slow envelope nudges fast phase. carrier_lag = np.empty_like(carrier) carrier_lag[:, 0] = carrier[:, 0] carrier_lag[:, 1:] = carrier[:, :-1] wobble = rng.uniform(0.0, 0.35, size=(n, 1)) * env out = amp * (carrier + wobble * carrier_lag) # Residual noise scaled by envelope intensity (prosody). sigma = rng.uniform(0.05, 0.35, size=(n, 1)) * (0.5 + 0.5 * np.abs(env)) phi = rng.uniform(0.0, 0.8, size=n) innov = rng.normal(0.0, 1.0, size=(n, L)) * sigma out = out + _ar1_batch(innov, phi) scale = np.exp(rng.uniform(np.log(0.2), np.log(40.0), size=(n, 1))) shift = rng.uniform(-50.0, 50.0, size=(n, 1)) return out * scale + shift def _delay_batch(x: np.ndarray, lags: np.ndarray) -> np.ndarray: """Causal delay with edge hold, vectorised over a small lag vocabulary. Rows sharing a lag are shifted in one slice copy — O(#unique_lags) passes instead of a Python loop over n. """ n, L = x.shape out = np.empty_like(x) # Hold initial value for the lag prefix. for lag in np.unique(lags): rows = np.nonzero(lags == lag)[0] if rows.size == 0: continue lag_i = int(lag) block = x[rows] delayed = np.empty_like(block) delayed[:, :lag_i] = block[:, :1] delayed[:, lag_i:] = block[:, :-lag_i] out[rows] = delayed return out def _got_causal(rng: np.random.Generator, n: int, L: int, *, lag_frac: float = 0.08) -> np.ndarray: """Production: Driver ▷ Response — lagged temporal causal chain. Inspired by Chronos-2 / CauKer temporal causal graphs, specialised to a univariate observable: the emitted series is a response driven by a latent driver with a drawn lag and FIR-like coupling, plus its own AR residual. Teaches lead-lag structure that pure mixture families never emit. """ lag_frac = float(np.clip(lag_frac, 0.01, 0.25)) # Discrete lag menu keeps _delay_batch on a handful of unique values. lag_menu = np.unique( np.clip( (np.array([0.01, 0.02, 0.04, 0.06, 0.08, 0.12, 0.16, 0.20]) * L).astype(np.int64), 1, max(1, int(L * lag_frac)), ) ) lags = rng.choice(lag_menu, size=n) driver = _sample_stem(rng, n, L) use_parent2 = rng.random(n) < 0.4 parent2 = _sample_stem(rng, n, L) a0 = rng.uniform(0.2, 1.2, size=(n, 1)) a1 = rng.uniform(0.3, 1.5, size=(n, 1)) b = rng.uniform(0.2, 1.0, size=(n, 1)) lags2 = rng.choice(lag_menu, size=n) delayed = _delay_batch(driver, lags) resp = a0 * driver + a1 * delayed if use_parent2.any(): delayed2 = _delay_batch(parent2, lags2) resp = np.where(use_parent2[:, None], resp + b * delayed2, resp) phi = rng.uniform(0.2, 0.9, size=n) sigma = rng.uniform(0.1, 0.5, size=(n, 1)) innov = rng.normal(0.0, 1.0, size=(n, L)) * sigma resp = resp + _ar1_batch(innov, phi) seas_on = rng.random((n, 1)) < 0.5 resp = resp + seas_on * _seasonal(rng, n, L, k_max=2) * rng.uniform( 0.1, 0.8, size=(n, 1) ) scale = np.exp(rng.uniform(np.log(0.2), np.log(40.0), size=(n, 1))) shift = rng.uniform(-50.0, 50.0, size=(n, 1)) return resp * scale + shift def _got_horizon( rng: np.random.Generator, n: int, L: int, *, horizon: int = 64, hi_frac: float = 0.25, exc_lo: float = 0.4, exc_hi: float = 2.5, clean_frac: float = 0.4, clean_lo: float = 0.02, clean_hi: float = 0.12, ) -> np.ndarray: """Production: Signal + short residual tuned to the eval forecast horizon. Cascade scores 4096-context → 64-step forecasts. This production makes that geometry explicit: a smooth, seasonally coherent signal that continues across the horizon, plus an AR residual whose correlation length is O(H) so noise averages inside the forecast window without erasing continuity. """ H = int(np.clip(horizon, 16, 256)) t = np.arange(L, dtype=np.float64)[None, :] / max(L - 1, 1) # Persistent multi-cadence signal (the forecastable backbone). signal = _seasonal(rng, n, L, k_max=3) # Mild spectral envelope so the signal is not pure sinusoids. mix_gp = rng.random(n) < 0.45 if mix_gp.any(): gp = _row_standardize(_spectral_gp(rng, n, L)) w = rng.uniform(0.2, 0.55, size=(n, 1)) signal = np.where(mix_gp[:, None], (1.0 - w) * signal + w * gp, signal) signal = _row_standardize(signal) _hi = rng.random((n, 1)) < hi_frac exc = np.where( _hi, rng.normal(0.0, exc_hi, size=(n, 1)), rng.normal(0.0, exc_lo, size=(n, 1)), ) signal = signal + exc * t # Residual with phi ~ exp(-1/H) so autocorr at lag H is ~e^{-1}. # Clean rows shrink residual further (sharp periodic reconstruction). phi_target = float(np.exp(-1.0 / H)) phi = rng.uniform(max(0.5, phi_target - 0.15), min(0.98, phi_target + 0.08), size=n) clean = rng.random((n, 1)) < clean_frac sigma = np.where( clean, rng.uniform(clean_lo, clean_hi, size=(n, 1)), rng.uniform(0.12, 0.55, size=(n, 1)), ) innov = rng.normal(0.0, 1.0, size=(n, L)) * sigma residual = _ar1_batch(innov, phi) # Sparse punctuation that recovers inside ~H steps (event + decay). impulses = _sparse_jumps(rng, n, L, rate=1.5 / L, scale=rng.uniform(1.0, 4.0, size=n)) recover_phi = rng.uniform(0.85, 0.98, size=n) events = _ar1_batch(impulses, recover_phi) use_events = rng.random(n) < 0.35 residual = residual + use_events[:, None] * events out = signal + residual scale = np.exp(rng.uniform(np.log(0.2), np.log(40.0), size=(n, 1))) shift = rng.uniform(-50.0, 50.0, size=(n, 1)) return out * scale + shift def _got_motif( rng: np.random.Generator, n: int, L: int, *, motif_max: int = 96, hi_frac: float = 0.25, exc_lo: float = 0.4, exc_hi: float = 2.5, ) -> np.ndarray: """Production: Tile(local shape) — non-sinusoidal repeating phrases. Pure Fourier seasonality under-covers weekday/shift/ops motifs that are shaped bumps, not sinusoids. Each row draws a short motif, tiles it across L, and applies slow amplitude/level drift so consecutive periods remain forecastable while still evolving. """ motif_max = int(np.clip(motif_max, 16, 256)) # Prefer periods near common cadences and the 64-step forecast window. period_menu = np.array( [7, 12, 16, 24, 32, 48, 64, 72, 96], dtype=np.int64 ) period_menu = period_menu[period_menu <= motif_max] periods = rng.choice(period_menu, size=n) t = np.arange(L, dtype=np.float64)[None, :] out = np.empty((n, L), dtype=np.float64) for p in np.unique(periods): rows = np.nonzero(periods == p)[0] m = int(rows.size) p_i = int(p) # Shape family: raised-cosine bump, asymmetric triangle, or AR snippet. kind = rng.integers(0, 3, size=m) motif = np.empty((m, p_i), dtype=np.float64) u = np.linspace(0.0, 1.0, p_i, endpoint=False)[None, :] cos_rows = kind == 0 if cos_rows.any(): width = rng.uniform(0.15, 0.55, size=(int(cos_rows.sum()), 1)) centre = rng.uniform(0.2, 0.8, size=(int(cos_rows.sum()), 1)) motif[cos_rows] = np.maximum( 0.0, np.cos(np.pi * (u - centre) / np.maximum(width, 1e-3)) ) tri_rows = kind == 1 if tri_rows.any(): peak = rng.uniform(0.2, 0.8, size=(int(tri_rows.sum()), 1)) left = np.clip(u / np.maximum(peak, 1e-3), 0.0, 1.0) right = np.clip((1.0 - u) / np.maximum(1.0 - peak, 1e-3), 0.0, 1.0) motif[tri_rows] = np.minimum(left, right) ar_rows = kind == 2 if ar_rows.any(): count = int(ar_rows.sum()) phi = rng.uniform(0.3, 0.9, size=count) innov = rng.normal(0.0, 1.0, size=(count, p_i)) motif[ar_rows] = _ar1_batch(innov, phi) motif = _row_standardize(motif) # Tile reps = int(np.ceil(L / p_i)) tiled = np.tile(motif, (1, reps))[:, :L] # Slow amplitude and level drift across tiles (forecastable evolution). n_tiles = max(1, int(np.ceil(L / p_i))) amp_path = np.cumsum( rng.normal(0.0, 0.08, size=(m, n_tiles)), axis=1 ) amp_path = 1.0 + 0.35 * _row_standardize(amp_path) level_path = np.cumsum( rng.normal(0.0, 0.05, size=(m, n_tiles)), axis=1 ) tile_idx = np.minimum(np.arange(L) // p_i, n_tiles - 1) amp = amp_path[:, tile_idx] level = level_path[:, tile_idx] # Within-period jitter so exact copies are rare. jitter = rng.normal(0.0, 0.05, size=(m, L)) out[rows] = amp * tiled + level + jitter _hi = rng.random((n, 1)) < hi_frac exc = np.where( _hi, rng.normal(0.0, exc_hi * 0.5, size=(n, 1)), rng.normal(0.0, exc_lo * 0.5, size=(n, 1)), ) tn = np.arange(L, dtype=np.float64)[None, :] / max(L - 1, 1) out = out + exc * tn # Light AR noise on top. phi = rng.uniform(0.0, 0.7, size=n) sigma = rng.uniform(0.05, 0.35, size=(n, 1)) out = out + _ar1_batch(rng.normal(0.0, 1.0, size=(n, L)) * sigma, phi) scale = np.exp(rng.uniform(np.log(0.2), np.log(40.0), size=(n, 1))) shift = rng.uniform(-50.0, 50.0, size=(n, 1)) return out * scale + shift # ── final safety gate ─────────────────────────────────────────────────────── def _sanitize(block: np.ndarray) -> np.ndarray: """Guarantee finite float64 values and proportionally bound each row. The trainer's ``check_series`` rejects any non-finite value, which would fail the whole run. Proportional rescaling preserves within-row geometry; hard clipping can create artificial constant plateaus on explosive paths. """ x = np.asarray(block, dtype=np.float64) np.nan_to_num(x, copy=False, nan=0.0, posinf=1e6, neginf=-1e6) if x.ndim == 1: peak = float(np.max(np.abs(x))) if peak > 1e6: x *= 1e6 / peak else: peak = np.max(np.abs(x), axis=1, keepdims=True) scale = np.where(peak > 1e6, 1e6 / np.maximum(peak, 1e-12), 1.0) x *= scale return x