"""zenfro_v6 — six-source blend aimed at 4096→64 forecast skill. Integrates complementary strengths from: v6 deltas vs v5: exact long-memory subtype, learnable pulse timing (periodic / seasonal / Hawkes), richer motif shapes, retail raised toward demand. Temporal grammar (informal CFG):: Series → Production | Stem | Domain Production → Compose | Splice | Nested | Causal | Horizon | Motif Domain → retail_demand | weather(light) Stem → {trend_seasonal, ar2, integrated, spectral, fGn/fBm, ou, …} 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 # Block size for the segmented AR(1) scan (~2*sqrt(L) Python iters vs L). _AR1_BLK = 32 # Multi-cadence seasonal bank (cascade-v21 coverage) with / # calendar bias: elevate 7 / 12 / 24 / 168 / 720 while retaining long-range # periods needed for 4096-context transfer. _SEASONAL_PERIODS = np.array( [4, 7, 12, 15, 24, 30, 48, 52, 60, 90, 96, 144, 168, 183, 240, 288, 336, 365, 672, 720, 730], dtype=np.float64, ) _SEASONAL_PROBS = np.array( [0.015, 0.12, 0.08, 0.025, 0.14, 0.025, 0.05, 0.025, 0.045, 0.015, 0.07, 0.04, 0.12, 0.015, 0.04, 0.035, 0.025, 0.03, 0.025, 0.05, 0.015], 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], [24, 720]], dtype=np.float64, ) # ── family mixture ────────────────────────────────────────────────────────── # Dynamics-heavy core + light domain families (retail, weather) + # GoT. Weather stays ~3.5% ( lesson). Retail raised toward 's 0.08. # Exact Davies–Harte long-memory + forecastable pulses are code-path upgrades # inside those families, not separate mixture slots. _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", "retail_demand", # / weekly demand "weather", # / archetypes, light mass "got_compose", "got_splice", "got_nested", "got_causal", "got_horizon", "got_motif", ) _DEFAULT_WEIGHTS: dict[str, float] = { # Dynamics-heavy core — primary 4096→64 transfer mass. "trend_seasonal_ar": 0.095, "regime_shift": 0.09, "multiplicative": 0.055, "ar2": 0.11, "integrated": 0.085, "threshold_ar": 0.035, "chaotic": 0.01, "spectral_gp": 0.05, "long_memory": 0.055, # +Davies–Harte exact fGn subtype (v17) "ou_stochastic_vol": 0.07, "physical_sensors": 0.01, "seasonal_counts": 0.015, "intermittent": 0.015, "pulse_outlier": 0.015, # forecastable event kinds (v17) # Domain families: demand emphasis, light weather (not 0.45). "retail_demand": 0.07, "weather": 0.035, # GoT slightly trimmed vs v5 to fund retail + exact long-memory. "got_compose": 0.045, "got_splice": 0.03, "got_nested": 0.03, "got_causal": 0.025, "got_horizon": 0.035, "got_motif": 0.02, } class Generator(DataGenerator): """Multi-source blend prior. 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-v6-six-blend")) 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, _retail_demand, _weather, 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) family = _FAMILIES[fam] # Preserve positivity for count/magnitude families and # exact integer/causal structure for count processes # ( family-aware artifacts). preserve_nonnegative = family in { "multiplicative", "physical_sensors", "seasonal_counts", "intermittent", "retail_demand", "weather", } if family == "retail_demand": preserve_integers: bool | np.ndarray = np.all( block == np.rint(block), axis=1 ) else: preserve_integers = family in { "seasonal_counts", "intermittent", } # Reverse only families whose laws remain valid under # time reversal. Indices: TSA, multiplicative, # spectral_gp, long_memory. allow_reverse = fam in (0, 2, 7, 8) # Prefix-calibrated hard bounds flatten random walks; # keep range artifacts off integrated paths. allow_range = family != "integrated" block = _sanitize( _measurement_artifacts( rng, block, preserve_nonnegative=preserve_nonnegative, preserve_integers=preserve_integers, allow_reverse=allow_reverse, allow_range_artifacts=allow_range, 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-v5-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, S: int = _AR1_BLK ) -> np.ndarray: """AR(1) filter ``x[:,t] = phi*x[:,t-1] + innov[:,t]`` along time. Segmented (block) scan from ````: ~S + L/S Python iterations instead of L, vectorised across the batch. Mathematically identical to the sequential recurrence for ``phi ∈ [0, 1)`` (FP relative error ~1e-13). Falls back to SciPy ``lfilter`` per row when any ``|phi| >= 1``. """ n, L = innov.shape p = np.asarray(phi, dtype=np.float64).reshape(n) if np.any(np.abs(p) >= 1.0): x = np.empty((n, L), dtype=np.float64) for i in range(n): x[i] = lfilter([1.0], [1.0, -float(p[i])], innov[i]) return x 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, :]).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_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 def _stationary_unit_ar1( rng: np.random.Generator, phi: np.ndarray, L: int ) -> np.ndarray: """Exact zero-mean, unit-variance stationary Gaussian AR(1) rows.""" p = np.asarray(phi, dtype=np.float64).reshape(-1) innov = rng.standard_normal((p.size, L)) * np.sqrt( np.maximum(1.0 - p[:, None] * p[:, None], 1e-12) ) innov[:, 0] = rng.standard_normal(p.size) return _ar1_batch(innov, p) def _prefix_mean_std( x: np.ndarray, *, calibration_points: int = 512 ) -> tuple[np.ndarray, np.ndarray]: """Location/scale from an initial calibration prefix only (causal).""" prefix = x[:, : min(x.shape[1], calibration_points)] mean = prefix.mean(axis=1, keepdims=True) std = prefix.std(axis=1, keepdims=True) return mean, np.where(std < 1e-12, 1.0, std) def _prefix_standardize( x: np.ndarray, *, center: bool = True, calibration_points: int = 512 ) -> np.ndarray: mean, std = _prefix_mean_std(x, calibration_points=calibration_points) return (x - mean) / std if center else x / std @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 _rfft_standard_normal( rng: np.random.Generator, n_rows: int, n_freq: int ) -> np.ndarray: """Gaussian coefficients for an even-length real inverse FFT. Interior complex bins have E|Z|²=1. DC and Nyquist are real N(0,1), as required by Hermitian symmetry. """ z = ( rng.standard_normal((n_rows, n_freq)) + 1j * rng.standard_normal((n_rows, n_freq)) ) / np.sqrt(2.0) z[:, 0] = rng.standard_normal(n_rows) z[:, -1] = rng.standard_normal(n_rows) return z def _normalized_logistic_curve( tn: np.ndarray, midpoint: np.ndarray, steepness: np.ndarray ) -> np.ndarray: """Monotone sigmoid curves normalized to an exact [0, 1] excursion.""" argument = np.clip(steepness * (tn - midpoint), -40.0, 40.0) raw = 1.0 / (1.0 + np.exp(-argument)) start = raw[:, :1] span = np.maximum(raw[:, -1:] - start, 1e-12) return (raw - start) / span def _sawtooth_wave( t: np.ndarray, period: np.ndarray, phase: np.ndarray, flipped: np.ndarray, ) -> np.ndarray: """Centered periodic ramps with an abrupt reset in either direction.""" cycle = np.mod(t / period + phase, 1.0) return np.where(flipped, 1.0 - cycle, cycle) - 0.5 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, preserve_integers: bool | np.ndarray = False, allow_reverse: bool = True, allow_range_artifacts: bool = True, 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. Rates are conservative. Sensor thresholds are calibrated from an initial observed prefix () so history does not depend on the unseen forecast target. ``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)) calibration_len = min(L, 512) reverse = (rng.random(n) < (0.06 * rs)) if allow_reverse else np.zeros(n, dtype=bool) out[reverse] = out[reverse, ::-1] if not preserve_nonnegative: invert = rng.random(n) < (0.04 * rs) out[invert] *= -1.0 # Draw selections even when range artifacts are disabled so RNG sequences # stay comparable across family-aware flags. for row in np.nonzero(rng.random(n) < (0.06 * rs))[0]: q = float(rng.uniform(0.03, 0.18)) upper = rng.random() < 0.5 if not allow_range_artifacts: continue calibration = out[row, :calibration_len] if upper: threshold = np.quantile(calibration, 1.0 - q) out[row] = np.minimum(out[row], threshold) else: threshold = np.quantile(calibration, q) out[row] = np.maximum(out[row], threshold) quantized = np.nonzero(rng.random(n) < (0.07 * rs))[0] if quantized.size: levels = rng.integers(16, 257, size=(quantized.size, 1)) if allow_range_artifacts: x = out[quantized] calibration = x[:, :calibration_len] lo = calibration.min(axis=1, keepdims=True) hi = calibration.max(axis=1, keepdims=True) step = (hi - lo) / np.maximum(levels - 1, 1) safe_step = np.where(step < 1e-12, 1.0, step) clipped = np.clip(x, lo, hi) out[quantized] = ( lo + np.rint((clipped - lo) / safe_step) * safe_step ) 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] if isinstance(preserve_integers, np.ndarray): integer_rows = np.asarray(preserve_integers, dtype=bool).reshape(n) out[integer_rows] = np.maximum(np.rint(out[integer_rows]), 0.0) elif preserve_integers: out = np.maximum(np.rint(out), 0.0) degenerate = out[:, :calibration_len].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 _davies_harte_fgn( rng: np.random.Generator, hurst: np.ndarray, L: int ) -> np.ndarray: """Exact fractional Gaussian noise via Davies–Harte embedding. Covariance γ(k)=0.5[(k+1)^(2H)-2k^(2H)+|k-1|^(2H)]. Embedding in a ``2L`` circulant gives a real Gaussian sample with the requested finite-lag covariance, unlike a generic ``1/f^β`` envelope. """ h = np.asarray(hurst, dtype=np.float64).reshape(-1, 1) n = h.shape[0] if n == 0: return np.empty((0, L), dtype=np.float64) k = np.arange(L, dtype=np.float64)[None, :] power = 2.0 * h covariance = 0.5 * ( (k + 1.0) ** power - 2.0 * k ** power + np.abs(k - 1.0) ** power ) circulant = np.concatenate( [covariance, np.zeros((n, 1)), covariance[:, 1:][:, ::-1]], axis=1 ) eigenvalues = np.maximum( np.fft.rfft(circulant, axis=1).real, 0.0 ) z = _rfft_standard_normal(rng, n, eigenvalues.shape[1]) return np.fft.irfft( z * np.sqrt(eigenvalues), n=2 * L, axis=1, norm="ortho", )[:, :L] def _long_memory(rng: np.random.Generator, n: int, L: int) -> np.ndarray: """Fractional power-law paths with both persistent and rough regimes. For Hurst H, fractional Gaussian noise has beta=2H-1. We sample that stationary increment process, then cumulatively sum selected rows to obtain mathematically consistent fractional Brownian motion paths. ~35% of rows use exact Davies–Harte fGn (); the rest keep the cheaper cascade9 spectral envelope on a 2L embedding so the evaluation target after a long context is not near an artificial wrap boundary. """ hurst = rng.uniform(0.3, 0.85, size=(n, 1)) level_path = rng.random((n, 1)) < 0.40 exact_core = rng.random(n) < 0.35 approximate_rows = np.nonzero(~exact_core)[0] exact_rows = np.nonzero(exact_core)[0] x = np.empty((n, L), dtype=np.float64) if approximate_rows.size: embed_len = 2 * L f = np.fft.rfftfreq(embed_len) safe_f = np.maximum(f, 1.0 / embed_len)[None, :] approximate_hurst = hurst[approximate_rows] beta = 2.0 * approximate_hurst - 1.0 amp = safe_f ** (-0.5 * beta) count = approximate_rows.size multiscale = rng.random((count, 1)) < 0.4 split_idx = rng.integers(8, max(9, f.size // 3), size=(count, 1)) split_f = np.maximum(split_idx / embed_len, 1.0 / embed_len) beta_hi = 2.0 * rng.uniform(0.3, 0.8, size=(count, 1)) - 1.0 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 = _rfft_standard_normal(rng, count, f.size) x[approximate_rows] = np.fft.irfft( z * amp, n=embed_len, axis=1, norm="ortho" )[:, :L] if exact_rows.size: x[exact_rows] = _davies_harte_fgn(rng, hurst[exact_rows], L) if np.any(level_path): level_rows = np.nonzero(level_path.reshape(-1))[0] x[level_rows] = np.cumsum(x[level_rows], axis=1) x[level_rows] -= x[level_rows, :1] return _prefix_standardize(x) 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 with weekly occurrence bias () # and a persistent latent occurrence/size state (). 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)) ) # Mild dedicated weekly modulation for retail-like domains. 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)) ) ) occurrence_state = _stationary_unit_ar1( rng, rng.uniform(0.0, 0.95, size=n), L ) occurrence_state = _prefix_standardize(occurrence_state) logit = ( np.log(base_p / (1.0 - base_p)) + season + np.log(np.clip(week, 0.2, 1.5)) + rng.uniform(0.0, 1.2, size=(n, 1)) * occurrence_state ) p = 1.0 / (1.0 + np.exp(-logit)) occur = (rng.random((n, L)) < p).astype(np.float64) independent_size_state = _stationary_unit_ar1( rng, rng.uniform(0.5, 0.98, size=n), L ) independent_size_state = _prefix_standardize(independent_size_state) coupling = rng.uniform(0.15, 0.55, size=(n, 1)) size_state = ( coupling * occurrence_state + np.sqrt(1.0 - coupling * coupling) * independent_size_state ) size_eta = rng.uniform(0.15, 0.45, size=(n, 1)) size_factor = np.exp(size_eta * size_state - 0.5 * size_eta * size_eta) magnitude = np.maximum(1.0, np.rint( 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) * size_factor )) return occur * magnitude def _retail_demand(rng: np.random.Generator, n: int, L: int) -> np.ndarray: """Weekly retail demand with persistent level drift and asymmetric events. From (qweqwe/dasadas demand lineage used by _v2). Each row has a seven-day profile, optional adjacent-day suppression, a slow latent log-demand walk, and short-lived promotions or stockouts. Mixes integer counts with continuous non-negative magnitude series. """ t = np.arange(L, dtype=np.float64)[None, :] tn = t / max(L - 1, 1) profile = rng.normal(0.0, 1.0, size=(n, 7)) profile -= profile.mean(axis=1, keepdims=True) special_days = rng.random(n) < 0.55 first_day = rng.integers(0, 7, size=n) depth = rng.uniform(0.4, 1.6, size=n) adjacent = np.zeros((n, 7), dtype=np.float64) rows = np.arange(n) adjacent[rows, first_day] -= depth adjacent[rows, (first_day + 1) % 7] -= depth adjacent -= adjacent.mean(axis=1, keepdims=True) profile += np.where(special_days[:, None], adjacent, 0.0) profile -= profile.mean(axis=1, keepdims=True) phase = rng.integers(0, 7, size=(n, 1)) day_index = (np.arange(L)[None, :] + phase) % 7 weekly = ( rng.uniform(0.03, 0.50, size=(n, 1)) * np.take_along_axis(profile, day_index, axis=1) ) excursion = ( rng.normal(0.0, 1.0, size=(n, 1)) * rng.uniform(0.3, 2.5, size=(n, 1)) ) latent_raw = np.cumsum( rng.normal(0.0, 1.0, size=(n, L)) * rng.uniform(0.005, 0.05, size=(n, 1)), axis=1, ) latent_walk = 3.0 * np.tanh(latent_raw / 3.0) promotions = np.zeros((n, L), dtype=np.float64) promo_rows, promo_cols = np.nonzero( rng.random((n, L)) < (rng.uniform(1.0, 8.0, size=(n, 1)) / L) ) promotions[promo_rows, promo_cols] = ( np.abs(rng.normal(0.0, 1.0, size=promo_rows.size)) * rng.uniform(0.5, 2.5, size=n)[promo_rows] ) promo_echo = np.zeros_like(promotions) promo_echo[:, 1:] = ( promotions[:, :-1] * rng.uniform(0.2, 0.6, size=(n, 1)) ) stockouts = np.zeros((n, L), dtype=np.float64) stock_rows, stock_cols = np.nonzero( rng.random((n, L)) < (rng.uniform(0.0, 4.0, size=(n, 1)) / L) ) stockouts[stock_rows, stock_cols] = ( np.abs(rng.normal(0.0, 1.0, size=stock_rows.size)) * rng.uniform(0.3, 1.5, size=n)[stock_rows] ) noise = ( rng.normal(0.0, 1.0, size=(n, L)) * rng.uniform(0.02, 0.25, size=(n, 1)) ) base_log_level = rng.uniform(np.log(0.2), np.log(3000.0), size=(n, 1)) log_mean = np.clip( base_log_level + excursion * tn + latent_walk + weekly + promotions + promo_echo - stockouts + noise, -8.0, 13.0, ) level = np.exp(log_mean) count_rows = rng.random(n) < 0.35 target_mean = np.exp( rng.uniform(np.log(1.0), np.log(80.0), size=(n, 1)) ) calibration_mean, _ = _prefix_mean_std(level) count_scale = target_mean / np.clip(calibration_mean, 1e-9, None) counts = rng.poisson( np.clip(level * count_scale, 0.0, 1.0e6) ).astype(np.float64) return np.where(count_rows[:, None], counts, level) def _cloud(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, :] phi_s = rng.uniform(0.990, 0.9990, size=n) syn = _ar1_batch(rng.normal(0.0, 1.0, (n, L)), phi_s) syn = (syn - syn.mean(1, keepdims=True)) / (syn.std(1, keepdims=True) + 1e-9) phi_m = rng.uniform(0.895, 0.95, size=n) mid = _ar1_batch(rng.normal(0.0, 1.0, (n, L)), phi_m) mid = (mid - mid.mean(1, keepdims=True)) / (mid.std(1, keepdims=True) + 1e-9) phi_f = rng.uniform(0.75, 0.87, size=n) fast = _ar1_batch(rng.normal(0.0, 1.0, (n, L)), phi_f) fast = (fast - fast.mean(1, keepdims=True)) / (fast.std(1, keepdims=True) + 1e-9) g = rng.normal(0.0, 1.0, (n, L)) spike = (rng.random((n, L)) < 0.05) * rng.uniform(5.0, 11.0, (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(1, keepdims=True) + 1e-9) diur = np.sin(2.0 * np.pi * t / 24.0 + rng.uniform(0.0, 2.0 * np.pi, (n, 1))) ws = 0.95 * rng.uniform(0.8, 1.2, (n, 1)) wm = 0.82 * rng.uniform(0.7, 1.3, (n, 1)) wf = 0.55 * rng.uniform(0.7, 1.3, (n, 1)) wr = 0.30 * rng.uniform(0.7, 1.3, (n, 1)) wd = 0.44 * rng.uniform(0.3, 1.4, (n, 1)) u = ws * syn + wm * mid + wf * fast + wr * rough + wd * diur centre = np.clip(rng.normal(61.0, 29.0, (n, 1)), 2.0, 98.0) span = 57.0 * rng.uniform(0.82, 1.18, (n, 1)) return np.round(np.clip(centre + span * u, 0.0, 100.0)) def _weather(rng: np.random.Generator, n: int, L: int) -> np.ndarray: """Sliced weather archetypes: diurnal base / night-floored radiation / cloud. From ```` with ```` quieter AR and optional monthly harmonic. Assigned first so expensive cloud red-noise runs only on ~18% of weather rows. Keep mixture weight low (~3.5%) for multi-domain pools. """ t = np.arange(L, dtype=np.float64)[None, :] rad_mask = rng.random(n) < 0.25 cloud_mask = (rng.random(n) < 0.24) & ~rad_mask wbp_mask = ~cloud_mask out = np.empty((n, L), dtype=np.float64) wbp_idx = np.nonzero(wbp_mask)[0] m = wbp_idx.size if m: radb = rad_mask[wbp_idx][:, 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) ) wk = (rng.random((m, 1)) < 0.45).astype(np.float64) seas += wk * 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)) ) month = (rng.random((m, 1)) < 0.35).astype(np.float64) seas += month * 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)) ) yamp = rng.uniform(0.25, 1.6, size=(m, 1)) yper = rng.uniform(2000.0, 9000.0, size=(m, 1)) yearly = yamp * np.sin( 2.0 * np.pi * t / yper + 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_batch(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.8, 1.6, size=(m, 1)) rad_diurnal = ramp * np.maximum( a1 * np.sin(2.0 * np.pi * t / 24.0 + p1) - thr, 0.0 ) night = rad_diurnal <= 0.0 rad_series = ( level + rad_diurnal + yearly + np.where(night, noise * 0.12, noise) ) out[wbp_idx] = np.where(radb, rad_series, base) cloud_idx = np.nonzero(cloud_mask)[0] if cloud_idx.size: out[cloud_idx] = _cloud(rng, int(cloud_idx.size), L) return out def _pulse_event_mask( rng: np.random.Generator, n: int, L: int ) -> tuple[np.ndarray, np.ndarray]: """Sample independent and history-dependent pulse occurrence processes. Kind 0 is a small calibration-only Poisson/Bernoulli branch. Kinds 1--3 carry forecastable timing through repeated cadence, seasonal conditional intensity, or self-excitation () — the structure that teaches the 64-step horizon when events are not pure iid shocks. """ kind = rng.choice(4, size=n, p=[0.15, 0.40, 0.35, 0.10]) events = np.zeros((n, L), dtype=bool) independent = np.nonzero(kind == 0)[0] if independent.size: rate = rng.uniform(2.0, 6.0, size=(independent.size, 1)) / max(L, 1) events[independent] = ( rng.random((independent.size, L)) < np.minimum(rate, 0.35) ) periodic = np.nonzero(kind == 1)[0] periods = rng.choice( np.asarray([24, 48, 96, 168, 256, 336, 512]), size=periodic.size, p=np.asarray([0.10, 0.15, 0.20, 0.20, 0.15, 0.10, 0.10]), ) phases = np.asarray( [rng.integers(0, max(int(period), 1)) for period in periods] ) for row, period, phase in zip(periodic, periods, phases, strict=True): nominal = np.arange(int(phase), L, int(period)) jitter = np.rint( rng.normal(0.0, max(1.0, 0.04 * period), size=nominal.size) ).astype(np.int64) starts = np.clip(nominal + jitter, 1, L - 1) events[row, starts] = True seasonal = np.nonzero(kind == 2)[0] if seasonal.size: t = np.arange(L, dtype=np.float64)[None, :] period = rng.choice( np.asarray([24.0, 48.0, 96.0, 168.0, 336.0]), size=(seasonal.size, 1), ) phase = rng.uniform(0.0, 2.0 * np.pi, size=(seasonal.size, 1)) base_rate = rng.uniform(4.0, 16.0, size=(seasonal.size, 1)) / max(L, 1) modulation = 0.15 + 1.70 * ( 0.5 + 0.5 * np.sin(2.0 * np.pi * t / period + phase) ) probability = np.minimum(base_rate * modulation, 0.35) events[seasonal] = rng.random((seasonal.size, L)) < probability # Discrete Hawkes analogue: p_t = mu + s_t, # s_{t+1} = decay*s_t + (1-decay)*branching*event_t. hawkes = np.nonzero(kind == 3)[0] if hawkes.size: baseline = rng.uniform(2.0, 8.0, size=hawkes.size) / max(L, 1) decay = rng.uniform(0.70, 0.96, size=hawkes.size) branching = rng.uniform(0.30, 0.80, size=hawkes.size) excitation = np.zeros(hawkes.size, dtype=np.float64) uniforms = rng.random((hawkes.size, L)) for step in range(L): occurred = uniforms[:, step] < np.minimum( baseline + excitation, 0.35 ) events[hawkes, step] = occurred excitation = ( decay * excitation + (1.0 - decay) * branching * occurred ) events[:, 0] = False return events, kind def _pulse_outlier(rng: np.random.Generator, n: int, L: int) -> np.ndarray: # Smooth base + forecastable event processes (periodic / seasonal / # self-exciting), persistent shock/recovery, 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)) events, kind = _pulse_event_mask(rng, n, L) magnitude_phi = rng.uniform(0.70, 0.98, size=n) magnitude_state = _stationary_unit_ar1(rng, magnitude_phi, L) magnitude_state = _prefix_standardize(magnitude_state) magnitude = rng.uniform(2.0, 8.0, size=(n, 1)) * np.exp( np.clip( rng.uniform(0.10, 0.40, size=(n, 1)) * magnitude_state, -1.0, 1.0, ) ) learnable_magnitude = ((kind == 1) | (kind == 2))[:, None] magnitude_cycle = 1.0 + 0.25 * np.sin( 2.0 * np.pi * np.arange(L, dtype=np.float64)[None, :] / rng.choice( np.asarray([96.0, 168.0, 336.0, 672.0]), size=(n, 1) ) + rng.uniform(0.0, 2.0 * np.pi, size=(n, 1)) ) magnitude *= np.where(learnable_magnitude, magnitude_cycle, 1.0) sign = rng.choice(np.asarray([-1.0, 1.0]), size=(n, 1)) impulses = events * sign * magnitude recovery = _ar1_batch(impulses, rng.uniform(0.75, 0.995, size=n)) sharp_shape = rng.random((n, 1)) < 0.45 series = base + np.where(sharp_shape, impulses, recovery) 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, triangle, AR, sawtooth, logistic (v17). kind = rng.integers(0, 5, 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) saw_rows = kind == 3 if saw_rows.any(): count = int(saw_rows.sum()) tt = np.arange(p_i, dtype=np.float64)[None, :] period = np.full((count, 1), float(p_i)) phase = rng.uniform(0.0, 1.0, size=(count, 1)) flipped = rng.random((count, 1)) < 0.5 motif[saw_rows] = _sawtooth_wave(tt, period, phase, flipped) log_rows = kind == 4 if log_rows.any(): count = int(log_rows.sum()) mid = rng.uniform(0.25, 0.75, size=(count, 1)) steep = rng.uniform(6.0, 24.0, size=(count, 1)) motif[log_rows] = _normalized_logistic_curve( np.broadcast_to(u, (count, p_i)), mid, steep ) 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