"""king-quality-v1 — cascade weekly-demand integration. This is the artifact a cascade miner actually competes with: a subclass of ``cascade.interface.DataGenerator`` that turns a single integer ``seed`` into a corpus of univariate float series. The subnet holds the model, seeds, and compute budget byte-identical between the king and every challenger, so the *only* thing that moves the forecast score is the distribution this file emits. The competitive lever is therefore **prior diversity + realism**: a corpus that covers more of the shapes a real forecaster must handle (trend, multi-seasonal, regime shifts, integrated/near-unit-root dynamics, smooth GP-like curves, nonlinear/chaotic recurrences, mean-reverting stochastic volatility, intermittent demand, event recovery, and measurement artifacts) trains a stronger zero-shot model than the reference generator's trend+seasonal+AR(1) mix. Design constraints this file respects (all from the contract in ``cascade.interface``): * **Determinism is load-bearing.** Every value is drawn from one ``np.random.default_rng(seed)`` in a fixed draw order, so two runs at the same seed produce byte-identical corpora — the property ``cascade verify`` audits by building the corpus twice and comparing digests. * **Code-only.** No shipped weights, no network, no clock, no un-seeded RNG. Imports stay on the dependency allowlist (NumPy/SciPy only) and clear of the static-guard blocklist. * **Bounded + finite.** Each series is 1-D ``(L,)`` float64, length in ``[min_length, max_length]``, finite (no NaN/inf). ``_sanitize`` is the last gate so a numerically unlucky draw can never poison a training run. Everything is **vectorised per family** (a batched time-axis recurrence, never a per-series Python loop over time). Compared with custom-fullctx-v4, the slow random-Fourier GP is replaced by FFT spectral sampling, a long-memory spectral family is added, and larger chunks amortise dispatch while staying far below the sandbox memory limit. """ 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 scipy.special import gammaln 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 = 2048 # Multi-cadence seasonal bank. Full 4096-point contexts can identify several # cycles even at 365/672/730-step periods, unlike short-crop generators. _SEASONAL_PERIODS = np.array( [4, 7, 12, 24, 30, 48, 52, 90, 96, 144, 168, 183, 288, 336, 365, 672, 730], dtype=np.float64, ) _SEASONAL_PROBS = np.array( [0.04, 0.12, 0.04, 0.16, 0.05, 0.06, 0.04, 0.03, 0.07, 0.03, 0.13, 0.04, 0.04, 0.06, 0.07, 0.04, 0.05], dtype=np.float64, ) _SEASONAL_PROBS /= _SEASONAL_PROBS.sum() # ── 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", # level + slope + multi-seasonal + AR(1) noise (rich reference) "regime_shift", # piecewise level/variance regimes with structural breaks "multiplicative", # positive level × seasonal factor × multiplicative noise "ar2", # AR(2), stationarity-guaranteed, incl. near-unit-root "integrated", # I(1)/I(2) random walks with drift "threshold_ar", # SETAR — regime-switching nonlinear recurrence "chaotic", # bounded chaotic maps (logistic / sine) "spectral_gp", # smooth GP-like paths via batched FFT sampling "long_memory", # persistent/anti-persistent power-law spectra "ou_stochastic_vol", # mean-reverting regimes + clustered/heavy-tailed volatility "physical_sensors", # bounded/skewed/smooth physical measurement archetypes "seasonal_counts", # seasonal Poisson/NB web and demand counts with bursts "intermittent", # zero-inflated / intermittent demand "pulse_outlier", # sharp/decaying events, outliers, and true flat runs "weekly_demand", # period-7 retail/web demand with promotions and dips ) # Dynamics-heavy composition selected by a controlled local A/B: it beat the # prior baseline on all three validation seeds (mean geomean 0.18431 vs # 0.19097). Config may still override these defaults, but copying generator.py # without config now retains the measured mixture. _DEFAULT_WEIGHTS: dict[str, float] = { # Preserve v16's relative composition at 92% and reserve the externally # validated sweet-spot weight for dedicated weekly demand. "trend_seasonal_ar": 0.1104, "regime_shift": 0.1104, "multiplicative": 0.0736, "ar2": 0.1380, "integrated": 0.1104, "threshold_ar": 0.0736, "chaotic": 0.0368, "spectral_gp": 0.0644, "long_memory": 0.0552, "ou_stochastic_vol": 0.0920, "physical_sensors": 0.0184, "seasonal_counts": 0.0184, "intermittent": 0.0092, "pulse_outlier": 0.0092, "weekly_demand": 0.08, } class Generator(DataGenerator): """A mixture-of-priors generator. 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)) @property def name(self) -> str: return str(self._cfg.get("name", "king-quality-v1")) 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, _integrated, _threshold_ar, _chaotic, _spectral_gp, _long_memory, _ou_stochastic_vol, _physical_sensors, _seasonal_counts, _intermittent, _pulse_outlier, _weekly_demand, ) # 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 and # exact integer/causal structure for count processes. preserve_nonnegative = fam in (2, 10, 11, 12, 14) preserve_integers = fam in (11, 12) block = _sanitize( _measurement_artifacts( rng, block, preserve_nonnegative=preserve_nonnegative, preserve_integers=preserve_integers, # Reverse only families whose laws remain valid # under time reversal. Reversing causal regimes, # SETAR maps, OU recovery, or shock paths creates # anti-causal precursors absent from the model. allow_reverse=fam in (0, 2, 7, 8), # A prefix-calibrated hard sensor bound turns an # unbounded random walk into an absorbing flat # line. Keep range artifacts on bounded families, # but preserve integrated dynamics. allow_range_artifacts=fam != 4, ) ) for row, series_i in enumerate(idx): length = int(lengths[series_i]) chunk[series_i] = np.ascontiguousarray( block[row, :length], dtype=np.float64 ) 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="cascade-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 def _prefix_mean_std( x: np.ndarray, *, calibration_points: int = 512 ) -> tuple[np.ndarray, np.ndarray]: """Location/scale estimated from an initial calibration prefix only. Whole-path normalization makes an emitted prefix depend on unseen future values and leaks the evaluation target into the synthetic process. A fixed early calibration interval keeps subsequent transformations 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) 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)[:, 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) jittered_period = per[modulated] * rng.uniform( 0.95, 1.05, size=(modulated.size, 1) ) modulated_arg = ( 2.0 * np.pi * t / jittered_period + 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 _measurement_artifacts( rng: np.random.Generator, block: np.ndarray, *, preserve_nonnegative: bool, preserve_integers: bool = False, allow_reverse: bool = True, allow_range_artifacts: bool = True, ) -> 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. """ original = np.asarray(block, dtype=np.float64) out = original.copy() n, L = out.shape reverse = (rng.random(n) < 0.06) if allow_reverse else np.zeros(n, dtype=bool) out[reverse] = out[reverse, ::-1] if not preserve_nonnegative: invert = rng.random(n) < 0.04 out[invert] *= -1.0 calibration_len = min(L, 512) # Calibrate sensor thresholds from the initial observed prefix. Using a # whole-path quantile would make history depend on the unseen target. # Selections and parameter draws happen even when range artifacts are # disabled, preserving the RNG sequence for controlled family comparisons. for row in np.nonzero(rng.random(n) < 0.06)[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)[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 ) # 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)[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 preserve_integers: out = np.maximum(np.rint(out), 0.0) # Heavy zero inflation plus upper censoring can otherwise collapse a sparse # row to its baseline. Such a row carries no forecasting signal. 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_shape = _prefix_standardize(seasonal_shape, center=False) seas = 1.0 + amp * seasonal_shape # ForecastPFN uses multiplicative Weibull noise so skew varies without # making signal-to-noise depend on the base level. Center on the exact # Weibull expectation to preserve trend and seasonality in expectation. shape = np.exp(rng.uniform(np.log(1.2), np.log(8.0), size=(n, 1))) uniform = np.maximum(rng.random((n, L)), np.finfo(np.float64).tiny) weibull = (-np.log(uniform)) ** (1.0 / shape) weibull_mean = np.exp(gammaln(1.0 + 1.0 / shape)) noise = 1.0 + rng.uniform(0.02, 0.15, size=(n, 1)) * ( weibull - weibull_mean ) 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)) burn = 512 innov = rng.normal(0.0, 1.0, size=(n, L + burn)) * sigma # Keep AR(2) genuinely stationary. The old 0.005-per-step drift accumulated # to roughly 20 units at full context and overwhelmed the AR covariance. return _ar2_batch(innov, a1, a2)[:, burn:] def _integrated(rng: np.random.Generator, n: int, L: int) -> np.ndarray: """Integrated paths with forecastable differenced dynamics. A corpus of pure iid random walks mostly teaches persistence because future increments are irreducible noise. Real integrated series more often have autocorrelated increments, recurring changes, or a persistent local drift. Retain an iid minority, but give most rows structure in first differences that a forecaster can identify from context. """ branch = rng.random(n) # Preserve a hard, genuine I(2) minority for coverage, but do not let it # dominate a family whose tiny local slope is poorly represented by the # fixed model's expanding global scaler. A separate near-I(2) branch uses # highly persistent (but stationary) velocity, retaining local-linear # forecast structure without cubic variance growth. order2 = branch < 0.10 persistent_velocity = (branch >= 0.10) & (branch < 0.45) drift = rng.normal(0.0, 0.02, size=(n, 1)) sigma = rng.uniform(0.2, 1.0, size=(n, 1)) raw = rng.normal(0.0, 1.0, size=(n, L)) * sigma # Correlated increments turn the family into a broad ARIMA prior instead of # almost exclusively an unpredictable random walk. Scaling innovations by # sqrt(1-phi²) keeps the marginal increment variance comparable across phi. correlated = (rng.random((n, 1)) < 0.70) | persistent_velocity[:, None] phi = rng.uniform(-0.35, 0.85, size=n) phi[persistent_velocity] = rng.uniform( 0.97, 0.999, size=persistent_velocity.sum() ) ar_steps = _ar1_batch( raw * np.sqrt(np.maximum(1.0 - phi[:, None] ** 2, 1e-3)), phi, ) steps = np.where(correlated, ar_steps, raw) # A minority has periodic first differences, as in seasonal ARIMA and # accumulated demand/sensor totals. Integrating a zero-mean periodic signal # remains bounded around the stochastic trend rather than exploding. seasonal_on = rng.random((n, 1)) < 0.30 seasonal_steps = _seasonal(rng, n, L, k_max=1) seasonal_steps = _prefix_standardize(seasonal_steps) steps += seasonal_on * seasonal_steps * sigma * rng.uniform(0.05, 0.35, size=(n, 1)) steps += drift walk = np.cumsum(steps, axis=1) walk2 = np.cumsum(walk, axis=1) o2 = order2[:, None] # I(2) variance grows cubically with length versus linearly for I(1). # Dividing by L aligns their standard-deviation order; sqrt(L) left the # double-integrated branch about sqrt(L) too large. return np.where(o2, walk2 / max(L, 1), 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)) burn = 256 total = L + burn innov = rng.normal(0.0, 1.0, size=(n, total)) * sigma x = np.empty((n, total), dtype=np.float64) x[:, 0] = innov[:, 0] for t in range(1, total): 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[:, burn:] 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) cur = x0.copy() # Remove initial-condition transients before emitting observations. for _ in range(64): nxt_log = r_log * cur * (1.0 - cur) nxt_sin = r_sin * np.sin(np.pi * cur) cur = np.clip(np.where(use_sine, nxt_sin, nxt_log), 0.0, 1.0) x = np.empty((n, L), dtype=np.float64) 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 x = _prefix_standardize(x) # Most real observations of nonlinear systems include measurement noise; # retain a clean minority to preserve the exact dynamical prior. noisy = rng.random((n, 1)) < 0.7 x += noisy * rng.normal(0.0, 1.0, size=(n, L)) * rng.uniform( 0.01, 0.15, size=(n, 1) ) return x * np.exp(rng.uniform(np.log(0.2), np.log(5.0), size=(n, 1))) \ + rng.normal(0.0, 2.0, size=(n, 1)) def _spectral_gp(rng: np.random.Generator, n: int, L: int) -> np.ndarray: """Composite RBF/Rational-Quadratic GP paths in O(n L log L). Chronos KernelSynth uses both kernels. We use a 2L circulant embedding and retain its first L samples. This preserves the requested kernel covariance over the emitted interval without making its two endpoints artificial neighbours, which an L-periodic inverse FFT would do. """ embed_len = 2 * L lag = np.minimum(np.arange(embed_len), embed_len - np.arange(embed_len))[None, :] lengthscale = np.exp(rng.uniform(np.log(8.0), np.log(256.0), size=(n, 1))) scaled_lag2 = (lag / lengthscale) ** 2 rbf_cov = np.exp(-0.5 * scaled_lag2) alpha = np.exp(rng.uniform(np.log(0.1), np.log(10.0), size=(n, 1))) rq_cov = (1.0 + scaled_lag2 / (2.0 * alpha)) ** (-alpha) blend = rng.beta(0.7, 0.7, size=(n, 1)) covariance = blend * rbf_cov + (1.0 - blend) * rq_cov spectrum = np.maximum(np.fft.rfft(covariance, axis=1).real, 0.0) z = rng.standard_normal(spectrum.shape) + 1j * rng.standard_normal(spectrum.shape) z[:, 0] = 0.0 x = np.fft.irfft(z * np.sqrt(spectrum), n=embed_len, axis=1)[:, :L] return _prefix_standardize(x) def _davies_harte_fgn( rng: np.random.Generator, hurst: np.ndarray, L: int ) -> np.ndarray: """Exact fractional Gaussian noise via Davies-Harte embedding. The covariance is ``γ(k)=0.5[(k+1)^(2H)-2k^(2H)+|k-1|^(2H)]``. Embedding it in a ``2L`` circulant matrix 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 = ( rng.standard_normal(eigenvalues.shape) + 1j * rng.standard_normal(eigenvalues.shape) ) / np.sqrt(2.0) z[:, 0] = rng.standard_normal(n) z[:, -1] = rng.standard_normal(n) 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. """ # Generate on a 2L embedding and retain only the first L samples. Directly # inverse-FFTing an L-point spectrum makes the emitted path circular, so an # evaluation target immediately after a long context sits near an # artificial wrap boundary. embed_len = 2 * L f = np.fft.rfftfreq(embed_len) safe_f = np.maximum(f, 1.0 / embed_len)[None, :] hurst = rng.uniform(0.3, 0.85, size=(n, 1)) level_path = rng.random((n, 1)) < 0.40 # Generate stationary fGn first (beta=2H-1), then integrate selected rows # to obtain actual non-stationary fBm. A steep beta=2H+1 spectrum sampled # directly by an inverse FFT is periodic coloured noise, not true fBm. beta = 2.0 * hurst - 1.0 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 / embed_len, 1.0 / embed_len) hurst_hi = rng.uniform(0.3, 0.8, size=(n, 1)) beta_hi = 2.0 * hurst_hi - 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 = rng.standard_normal((n, f.size)) + 1j * rng.standard_normal((n, f.size)) x = np.fft.irfft(z * amp, n=embed_len, axis=1)[:, :L] # Most rows use exact fGn covariance. Retain a minority of the approximate # piecewise-spectrum paths because scale-dependent roughness is useful # real-world coverage not represented by a single Hurst exponent. exact_core = rng.random(n) < 0.65 exact_rows = np.nonzero(exact_core)[0] if exact_rows.size: x[exact_rows] = _davies_harte_fgn( rng, hurst[exact_rows], L ) # fBm is the cumulative sum of fGn. Remove the initial level so scale and # shift augmentation elsewhere do not depend on an arbitrary FFT endpoint. 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. # Most regimes persist long enough to infer from context. Retain a small # rapid-switch minority rather than making the synthetic task uniformly # easier and deleting realistic hard cases. rapid_switch = rng.random((n, 1)) < 0.15 persistent_rate = np.exp( rng.uniform(np.log(0.0005), np.log(0.03), size=(n, 1)) ) rapid_rate = np.exp(rng.uniform(np.log(0.03), np.log(0.15), size=(n, 1))) switch_rate = np.where(rapid_switch, rapid_rate, persistent_rate) 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. speed = rng.random((n, 1)) ultra_slow = speed < 0.20 slow = (speed >= 0.20) & (speed < 0.65) fast_phi = rng.uniform(0.900, 0.980, size=(n, 1)) slow_phi = rng.uniform(0.985, 0.9975, size=(n, 1)) ultra_slow_phi = rng.uniform(0.9975, 0.9995, size=(n, 1)) phi = np.where( ultra_slow, ultra_slow_phi, np.where(slow, slow_phi, fast_phi), ) 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 sigma_seasonal_on = rng.random((n, 1)) < 0.3 seasonal_component = np.zeros((n, L), dtype=np.float64) seasonal_rows = np.nonzero( (seasonal_on | sigma_seasonal_on).reshape(-1) )[0] if seasonal_rows.size: seasonal_component[seasonal_rows] = _seasonal( rng, int(seasonal_rows.size), L, k_max=3 ) mean += seasonal_on * seasonal_component \ * rng.uniform(0.5, 3.0, size=(n, 1)) # Mean-reverting log volatility gives clustered but bounded uncertainty. # The persistence range maps TempoPFN's kappa_v=[0.5,5] at dt=0.01 into # exp(-kappa_v*dt)≈[0.951,0.995]. log_sigma0 = rng.normal(np.log(0.3), 0.3, size=(n, 1)) log_sigma1 = rng.normal(np.log(1.5), 0.5, size=(n, 1)) log_sigma_mean = np.where(regime == 0, log_sigma0, log_sigma1) vol_rho = rng.uniform(0.951, 0.995, size=(n, 1)) vol_eta = rng.uniform(0.03, 0.20, size=(n, 1)) vol_eps = rng.standard_normal((n, L)) vol_drive = (1.0 - vol_rho) * log_sigma_mean \ + np.sqrt(1.0 - vol_rho * vol_rho) * vol_eta * vol_eps log_vol = np.empty((n, L), dtype=np.float64) log_vol[:, 0] = log_sigma_mean[:, 0] for i in range(n): rho = float(vol_rho[i, 0]) log_vol[i, 1:] = lfilter( [1.0], [1.0, -rho], vol_drive[i, 1:], zi=[rho * log_vol[i, 0]], )[0] vol = np.exp(np.clip(log_vol, -5.0, 5.0)) # TempoPFN independently applies seasonality to sigma in 30% of paths. sigma_seasonal = sigma_seasonal_on * seasonal_component * rng.uniform( 0.03, 0.18, size=(n, 1) ) vol *= np.exp(np.clip(sigma_seasonal, -0.7, 0.7)) 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()) # Sample a per-step diffusion coefficient. Dividing by sqrt(L) made the # process depend on the requested total length, so prefixes generated # under different horizons did not share one stochastic law. diffusion = np.exp( rng.uniform(np.log(0.03), np.log(0.20), size=(count, 1)) ) walk = np.cumsum( rng.standard_normal((count, L)) * diffusion, axis=1 ) level = rng.uniform(900.0, 1100.0, size=(count, 1)) out[pressure] = level + 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 sampled_day_period = rng.choice([24, 48, 96, 144], size=(n, 1)) # A seven-step primary period represents daily observations, so each sample # is one day. Sub-daily rows retain a cadence-scaled day length. day_period = np.where(period <= 7.0, 1, sampled_day_period) 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. A separate # lognormal-AR intensity branch makes overdispersion persistent through # time rather than redrawing an unrelated multiplier at every step. overdispersion_kind = rng.random((n, 1)) gamma_mixed = overdispersion_kind < 0.35 persistent_mixed = (overdispersion_kind >= 0.35) & ( overdispersion_kind < 0.70 ) shape = rng.uniform(0.5, 4.0, size=(n, 1)) gamma_intensity = lam * rng.gamma(shape, 1.0 / shape, size=(n, L)) intensity_state = _ar1_batch( rng.standard_normal((n, L)), rng.uniform(0.70, 0.995, size=n), ) intensity_state = _prefix_standardize(intensity_state) eta = rng.uniform(0.10, 0.60, size=(n, 1)) # Center the lognormal multiplier at expectation one. persistent_intensity = lam * np.exp( np.clip(eta * intensity_state - 0.5 * eta * eta, -3.0, 3.0) ) mixed = np.where( gamma_mixed, gamma_intensity, np.where(persistent_mixed, persistent_intensity, lam), ) return rng.poisson(mixed).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)) ) # Intermittent-demand models separate occurrence from positive size. A # persistent latent state makes occurrence probability evolve over time # instead of producing iid Bernoulli zeros. occurrence_state = _ar1_batch( rng.normal(0.0, 1.0, size=(n, L)), rng.uniform(0.0, 0.95, size=n), ) occurrence_state = _prefix_standardize(occurrence_state) logit = np.log(base_p / (1.0 - base_p)) + season \ + 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) # Positive demand sizes are rarely iid in practice: customer/product scale # and local demand intensity persist. Couple a smooth latent size state # weakly to the occurrence state, while retaining gamma observation noise. independent_size_state = _ar1_batch( rng.normal(0.0, 1.0, size=(n, L)), rng.uniform(0.5, 0.98, size=n), ) 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_factor = np.exp(np.clip( rng.uniform(0.15, 0.45, size=(n, 1)) * size_state, -1.5, 1.5 )) 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 _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 information through repeated cadence, seasonal conditional intensity, or self-excitation respectively. """ kind = rng.choice(4, size=n, p=[0.15, 0.35, 0.30, 0.20]) events = np.zeros((n, L), dtype=bool) # Independent innovations: history identifies only the marginal event rate, # never the exact next event. Keep this branch small but nonzero so quantile # forecasts still learn honest tail mass. 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)) < rate # Repeated events with modest timing jitter, following the learnable # periodic/clustered spike construction used by synthetic forecasting # priors. At least several cycles occur in a full-context series. 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 # A cyclic conditional intensity makes event probability forecastable while # retaining irreducible Bernoulli timing uncertainty. 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) ) events[seasonal] = ( rng.random((seasonal.size, L)) < base_rate * modulation ) # Discrete Hawkes analogue: # p_t = mu + s_t, # s_{t+1} = decay*s_t + (1-decay)*branching*event_t. # Its expected offspring count is ``branching < 1``, so it is stable, and # observed events raise the near-future conditional event probability. 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: # A smooth base with isolated innovations, predictable event processes, # 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)) events, kind = _pulse_event_mask(rng, n, L) magnitude_phi = rng.uniform(0.70, 0.98, size=n) magnitude_state = _ar1_batch( rng.normal(0.0, 1.0, size=(n, L)) * np.sqrt(1.0 - magnitude_phi[:, None] ** 2), magnitude_phi, ) 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, ) ) # Periodic and seasonal rows also receive slowly evolving event magnitude; # independent pulses retain iid timing and Hawkes rows derive predictability # from occurrence clustering rather than a fabricated deterministic trend. 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) # 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 def _weekly_demand( rng: np.random.Generator, n: int, L: int ) -> np.ndarray: """Non-negative period-7 demand with promotions, dips, and count rows. Adapted from the public ``j-test/dasadas`` generator. Its dedicated 8% demand prior beat that generator's base model on a multi-domain pool, while 10% overshot. In king-quality-v1 the useful process is integrated without replacing cascade-v16's richer GP, long-memory, OU, sensor, and count families. """ time = np.arange(L, dtype=np.float64)[None, :] normalized_time = time / max(L - 1, 1) # A learned day-of-week profile, with a weekend-dip branch whose two-day # phase is randomized so it remains a generic weekly demand prior. seasonal_amplitude = rng.uniform(0.03, 0.5, size=(n, 1)) profile = rng.normal(0.0, 1.0, size=(n, 7)) profile -= profile.mean(axis=1, keepdims=True) has_weekend_dip = rng.random(n) < 0.5 dip_start = rng.integers(0, 7, size=n) dip_depth = rng.uniform(0.4, 1.6, size=n) weekend_profile = np.zeros((n, 7), dtype=np.float64) rows = np.arange(n) weekend_profile[rows, dip_start] -= dip_depth weekend_profile[rows, (dip_start + 1) % 7] -= dip_depth weekend_profile -= weekend_profile.mean(axis=1, keepdims=True) profile += np.where( has_weekend_dip[:, None], weekend_profile, 0.0 ) profile -= profile.mean(axis=1, keepdims=True) phase = rng.integers(0, 7, size=(n, 1)) weekday_index = (np.arange(L)[None, :] + phase) % 7 weekly_log = seasonal_amplitude * np.take_along_axis( profile, weekday_index, axis=1 ) excursion = ( rng.normal(0.0, 1.0, size=(n, 1)) * rng.uniform(0.3, 2.5, size=(n, 1)) ) trend = excursion * normalized_time step_scale = rng.uniform(0.005, 0.05, size=(n, 1)) random_walk = np.clip( np.cumsum( rng.normal(0.0, 1.0, size=(n, L)) * step_scale, axis=1, ), -3.0, 3.0, ) # Sparse promotions have a one-step echo; independent negative events # represent holidays, outages, or temporary stock constraints. promotion_mask = rng.random((n, L)) < ( rng.uniform(1.0, 8.0, size=(n, 1)) / L ) promotions = ( promotion_mask * np.abs(rng.normal(0.0, 1.0, size=(n, L))) * rng.uniform(0.5, 2.5, size=(n, 1)) ) echo = np.zeros_like(promotions) echo[:, 1:] = ( promotions[:, :-1] * rng.uniform(0.2, 0.6, size=(n, 1)) ) promotions += echo holiday_mask = rng.random((n, L)) < ( rng.uniform(0.0, 4.0, size=(n, 1)) / L ) holiday_dips = ( holiday_mask * np.abs(rng.normal(0.0, 1.0, size=(n, L))) * rng.uniform(0.3, 1.5, size=(n, 1)) ) noise = ( rng.normal(0.0, 1.0, size=(n, L)) * rng.uniform(0.02, 0.25, size=(n, 1)) ) base = rng.uniform(0.0, 8.0, size=(n, 1)) log_mean = np.clip( base + trend + random_walk + weekly_log + promotions - holiday_dips + noise, -8.0, 13.0, ) level = np.exp(log_mean) # A minority is emitted as exact Poisson demand; the rest remains positive # continuous magnitude data such as traffic, revenue, or energy load. is_count = rng.random(n) < 0.35 count_scale = rng.uniform(1.0, 60.0, size=(n, 1)) / np.clip( level.mean(axis=1, keepdims=True), 1e-9, None ) counts = rng.poisson( np.clip(level * count_scale, 0.0, 1e6) ).astype(np.float64) return np.where(is_count[:, None], counts, level) # ── final safety gate ─────────────────────────────────────────────────────── def _sanitize(block: np.ndarray) -> np.ndarray: """Guarantee the contract: finite float64, no NaN/inf, bounded magnitude. The trainer's ``check_series`` rejects any non-finite value, which would fail the whole run — so this is the hard backstop after every family builder. Replaces non-finite values and clips to a generous bound. """ x = np.asarray(block, dtype=np.float64) np.nan_to_num(x, copy=False, nan=0.0, posinf=1e6, neginf=-1e6) np.clip(x, -1e6, 1e6, out=x) return x