"""goethe""" 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 os import numpy as np from numba import njit, prange, set_num_threads from cascade.interface import DataGenerator # goethe # goethe # goethe # goethe set_num_threads(max(1, (os.cpu_count() or 4) // 2)) # goethe # goethe _CHUNK = 2048 # goethe # goethe _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() # goethe _FAMILIES: tuple[str, ...] = ( "trend_seasonal_ar", # goethe "regime_shift", # goethe "multiplicative", # goethe "ar2", # goethe "integrated", # goethe "threshold_ar", # goethe "chaotic", # goethe "spectral_gp", # goethe "long_memory", # goethe "ou_stochastic_vol", # goethe "physical_sensors", # goethe "seasonal_counts", # goethe "intermittent", # goethe "pulse_outlier", # goethe ) _DEFAULT_WEIGHTS: dict[str, float] = { "trend_seasonal_ar": 0.12, "regime_shift": 0.12, "multiplicative": 0.08, "ar2": 0.15, "integrated": 0.12, "threshold_ar": 0.08, "chaotic": 0.04, "spectral_gp": 0.07, "long_memory": 0.06, "ou_stochastic_vol": 0.10, "physical_sensors": 0.02, "seasonal_counts": 0.02, "intermittent": 0.01, "pulse_outlier": 0.01, } # goethe # goethe # goethe # goethe # goethe @njit(cache=False, parallel=True, fastmath=True) def _ar1_recur_jit(drive: np.ndarray, phi: np.ndarray, x0: np.ndarray) -> np.ndarray: """goethe""" n, L = drive.shape x = np.empty((n, L), dtype=np.float64) for i in prange(n): p = phi[i] cur = x0[i] x[i, 0] = cur for t in range(1, L): cur = p * cur + drive[i, t] x[i, t] = cur return x @njit(cache=False, parallel=True, fastmath=True) def _ar2_batch_jit(innov: np.ndarray, a1: np.ndarray, a2: np.ndarray) -> np.ndarray: n, L = innov.shape x = np.empty((n, L), dtype=np.float64) for i in prange(n): aa1 = a1[i] aa2 = a2[i] x[i, 0] = innov[i, 0] if L > 1: x[i, 1] = aa1 * x[i, 0] + innov[i, 1] for t in range(2, L): x[i, t] = aa1 * x[i, t - 1] + aa2 * x[i, t - 2] + innov[i, t] return x @njit(cache=False, parallel=True, fastmath=True) def _threshold_ar_jit( innov: np.ndarray, phi_hi: np.ndarray, phi_lo: np.ndarray, const_hi: np.ndarray, const_lo: np.ndarray, ) -> np.ndarray: n, L = innov.shape x = np.empty((n, L), dtype=np.float64) for i in prange(n): x[i, 0] = innov[i, 0] for t in range(1, L): prev = x[i, t - 1] if prev >= 0.0: phi = phi_hi[i] const = const_hi[i] else: phi = phi_lo[i] const = const_lo[i] v = const + phi * prev + innov[i, t] if v > 1e6: v = 1e6 elif v < -1e6: v = -1e6 x[i, t] = v return x @njit(cache=False, parallel=True, fastmath=True) def _chaotic_jit(map_id: np.ndarray, r_a: np.ndarray, x0: np.ndarray, L: int) -> np.ndarray: """goethe""" n = x0.shape[0] x = np.empty((n, L), dtype=np.float64) for i in prange(n): cur = x0[i] x[i, 0] = cur mid = map_id[i] a = r_a[i] for t in range(1, L): if mid == 1: cur = a * np.sin(np.pi * cur) elif mid == 2: if cur < 0.5: cur = a * cur else: cur = a * (1.0 - cur) else: cur = a * cur * (1.0 - cur) if cur < 0.0: cur = 0.0 elif cur > 1.0: cur = 1.0 x[i, t] = cur return x @njit(cache=False, parallel=True, fastmath=True) def _apply_hold_runs_jit( series: np.ndarray, row_ptr: np.ndarray, cols: np.ndarray, run_lengths: np.ndarray, ) -> None: """Freeze variable-length runs at their pre-event value, in place. ``cols``/``run_lengths`` are grouped by row via the CSR-style ``row_ptr`` (built from ``np.nonzero`` on a 2D mask, which is already row-major and column-ascending within a row). Parallelizing over rows (``prange``) is therefore safe: each row's events are still applied in increasing-time order (so a later hold can start from an already-held value, matching the original per-row sequential semantics) while independent rows run concurrently. """ n, L = series.shape for r in prange(n): lo = row_ptr[r] hi = row_ptr[r + 1] for k in range(lo, hi): c = cols[k] run = run_lengths[k] end = c + run if end > L: end = L val = series[r, c - 1] for t in range(c, end): series[r, t] = val @njit(cache=False, parallel=False, fastmath=True) def _seasonal_add_stationary_jit( out: np.ndarray, rows: np.ndarray, sin_basis: np.ndarray, cos_basis: np.ndarray, basis_idx: np.ndarray, amp: np.ndarray, phase: np.ndarray, ) -> None: """Add stationary sinusoids from the cached cadence bank into ``out``. Serial on purpose: at typical per-family batch sizes, parallel prange was slower than a tight serial loop due to thread-pool overhead. """ L = out.shape[1] for i in range(rows.shape[0]): r = rows[i] bi = basis_idx[i] am = amp[r] ph = phase[r] cph = np.cos(ph) sph = np.sin(ph) for t in range(L): out[r, t] += am * (sin_basis[bi, t] * cph + cos_basis[bi, t] * sph) @njit(cache=False, parallel=False, fastmath=True) def _seasonal_add_modulated_jit( out: np.ndarray, rows: np.ndarray, per: np.ndarray, amp: np.ndarray, phase: np.ndarray, m_per: np.ndarray, m_phase: np.ndarray, amp_scale: np.ndarray, phase_scale: np.ndarray, ) -> None: """Add amplitude/phase-modulated sinusoids into ``out`` (no basis reuse).""" L = out.shape[1] two_pi = 2.0 * np.pi for i in range(rows.shape[0]): r = rows[i] p = per[i] am = amp[i] ph = phase[i] mp = m_per[i] mph = m_phase[i] as_ = amp_scale[i] ps = phase_scale[i] inv_p = two_pi / p inv_mp = two_pi / mp inv_mp17 = two_pi / (1.7 * mp) for t in range(L): tf = float(t) slow = np.sin(inv_mp * tf + mph) amp_mod = 1.0 + as_ * slow phase_mod = ps * np.sin(inv_mp17 * tf - mph) out[r, t] += am * amp_mod * np.sin(inv_p * tf + ph + phase_mod) class Generator(DataGenerator): """goethe""" 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)) 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() 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)) # goethe self._seasonal_focus = float(np.clip(cfg.get("seasonal_focus", 0.75), 0.0, 1.0)) # goethe # goethe # goethe # goethe self._het_noise_frac = float(np.clip(cfg.get("het_noise_frac", 0.35), 0.0, 1.0)) # goethe # goethe # goethe self._ar2_seasonal_frac = float(np.clip(cfg.get("ar2_seasonal_frac", 0.25), 0.0, 1.0)) self._integrated_seasonal_frac = float( np.clip(cfg.get("integrated_seasonal_frac", 0.30), 0.0, 1.0) ) self._fixed_len = self._min_len == self._max_len @property def name(self) -> str: return str(self._cfg.get("name", "cascade-fullctx-persist-v13")) def generate(self, n_series: int) -> Iterator[np.ndarray]: if n_series <= 0: return # goethe # goethe # goethe _warmup_kernels(self._max_len) rng = np.random.default_rng(self._seed) max_len = self._max_len fixed_len = self._fixed_len period_p = (1.0 - self._seasonal_focus) * (np.ones_like(_SEASONAL_PROBS) / len(_SEASONAL_PROBS)) period_p = period_p + self._seasonal_focus * _SEASONAL_PROBS period_p = period_p / period_p.sum() 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, het_frac=self._het_noise_frac, period_p=period_p, ), partial(_regime_shift, period_p=period_p), partial( _multiplicative, hi_frac=self._tr_hi_frac, exc_lo=self._gr_exc_lo, exc_hi=self._gr_exc_hi, period_p=period_p, ), partial(_ar2, period_p=period_p, seasonal_frac=self._ar2_seasonal_frac), partial(_integrated, period_p=period_p, seasonal_frac=self._integrated_seasonal_frac), _threshold_ar, _chaotic, _spectral_gp, _long_memory, partial(_ou_stochastic_vol, period_p=period_p), partial(_physical_sensors, period_p=period_p), partial(_seasonal_counts, period_p=period_p), _intermittent, partial(_pulse_outlier, period_p=period_p), ) # goethe # goethe 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(): # goethe # goethe 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) # goethe preserve_nonnegative = fam in (2, 10, 11, 12) block = _sanitize( _measurement_artifacts( rng, block, preserve_nonnegative=preserve_nonnegative ) ) # goethe # goethe # goethe if fixed_len: for row, series_i in enumerate(idx): chunk[int(series_i)] = block[row].copy() else: for row, series_i in enumerate(idx): length = int(lengths[series_i]) chunk[int(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: # goethe 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]: if arr is None: # goethe raise RuntimeError("internal: unfilled series slot") yield arr finally: stop.set() producer.join(timeout=1.0) # goethe def _ar1_batch(innov: np.ndarray, phi: np.ndarray) -> np.ndarray: """goethe""" return _ar1_recur(innov, phi, innov[:, 0]) def _ar1_recur(drive: np.ndarray, phi: np.ndarray, x0: np.ndarray) -> np.ndarray: return _ar1_recur_jit( np.ascontiguousarray(drive, dtype=np.float64), np.ascontiguousarray(phi.reshape(-1), dtype=np.float64), np.ascontiguousarray(x0.reshape(-1), dtype=np.float64), ) def _ar2_batch(innov: np.ndarray, a1: np.ndarray, a2: np.ndarray) -> np.ndarray: return _ar2_batch_jit( np.ascontiguousarray(innov), np.ascontiguousarray(a1.reshape(-1)), np.ascontiguousarray(a2.reshape(-1)), ) @lru_cache(maxsize=4) def _seasonal_basis(L: int) -> tuple[np.ndarray, np.ndarray]: """goethe""" angle = ( 2.0 * np.pi * np.arange(L, dtype=np.float64)[None, :] / _SEASONAL_PERIODS[:, None] ) return np.sin(angle), np.cos(angle) @lru_cache(maxsize=4) def _rfftfreq_cached(L: int) -> np.ndarray: """goethe""" return np.fft.rfftfreq(L) _WARMUP_DONE = False def _warmup_kernels(L: int) -> None: """goethe""" global _WARMUP_DONE if _WARMUP_DONE: return n, Lw = 4, min(int(L), 64) innov = np.ascontiguousarray(np.zeros((n, Lw), dtype=np.float64)) phi = np.zeros(n, dtype=np.float64) _ar1_recur_jit(innov, phi, innov[:, 0].copy()) _ar2_batch_jit(innov, phi, phi) _threshold_ar_jit(innov, phi, phi, phi, phi) _chaotic_jit(np.zeros(n, dtype=np.int64), phi, phi + 0.5, Lw) _apply_hold_runs_jit( innov.copy(), np.zeros(n + 1, dtype=np.int64), np.empty(0, dtype=np.int64), np.empty(0, dtype=np.int64), ) sin_b, cos_b = _seasonal_basis(Lw) rows = np.arange(n, dtype=np.int64) _seasonal_add_stationary_jit( innov.copy(), rows, sin_b, cos_b, np.zeros(n, dtype=np.int64), phi + 1.0, phi, ) _seasonal_add_modulated_jit( innov.copy(), rows, np.full(n, 24.0), phi + 1.0, phi, np.full(n, 96.0), phi, phi + 0.1, phi + 0.1, ) _rfftfreq_cached(Lw) _WARMUP_DONE = True def _seasonal( rng: np.random.Generator, n: int, L: int, k_max: int = 3, period_p: np.ndarray | None = None, ) -> np.ndarray: """Sum of 1..k_max stationary or slowly modulated seasonal components. RNG draws stay in NumPy (fixed draw order). Assembly of the (n, L) output runs in compiled kernels: stationary rows reuse the cached cadence bank; modulated rows skip the wasted stationary compute they used to overwrite. """ sin_basis, cos_basis = _seasonal_basis(L) p = _SEASONAL_PROBS if period_p is None else period_p k = rng.integers(1, k_max + 1, size=n) out = np.zeros((n, L), dtype=np.float64) n_periods = _SEASONAL_PERIODS.shape[0] for j in range(k_max): # goethe basis_idx_all = rng.choice(n_periods, size=n, p=p) per = _SEASONAL_PERIODS[basis_idx_all] amp = rng.uniform(0.2, 2.0, size=n) phase = rng.uniform(0.0, 2.0 * np.pi, size=n) mod_flag = rng.random(n) < 0.35 active = k > j modulated = np.nonzero(active & mod_flag)[0] stationary = np.nonzero(active & ~mod_flag)[0] if stationary.size: _seasonal_add_stationary_jit( out, stationary.astype(np.int64, copy=False), sin_basis, cos_basis, basis_idx_all[stationary].astype(np.int64, copy=False), amp, phase, ) if modulated.size: # goethe # goethe # goethe m_per = np.clip( per[modulated] * rng.uniform(4.0, 12.0, size=modulated.size), 32.0, 2.0 * L, ) m_phase = rng.uniform(0.0, 2.0 * np.pi, size=modulated.size) amp_scale = rng.uniform(0.05, 0.45, size=modulated.size) phase_scale = rng.uniform(0.05, 0.75, size=modulated.size) _seasonal_add_modulated_jit( out, modulated.astype(np.int64, copy=False), per[modulated], amp[modulated], phase[modulated], m_per, m_phase, amp_scale, phase_scale, ) return out def _sparse_jumps(rng: np.random.Generator, n: int, L: int, rate: float, scale) -> np.ndarray: """goethe""" 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 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 _apply_hold_runs(rng: np.random.Generator, series: np.ndarray, rate: float) -> None: """goethe""" n, L = series.shape starts = rng.random((n, L)) < rate starts[:, 0] = False rows, cols = np.nonzero(starts) if rows.size == 0: return run_lengths = rng.integers(3, 65, size=rows.size) row_ptr = np.searchsorted(rows, np.arange(n + 1)).astype(np.int64) _apply_hold_runs_jit( series, row_ptr, np.ascontiguousarray(cols, dtype=np.int64), np.ascontiguousarray(run_lengths, dtype=np.int64), ) def _measurement_artifacts( rng: np.random.Generator, block: np.ndarray, *, preserve_nonnegative: bool, ) -> np.ndarray: """Apply sparse, cheap real-measurement effects to a generated block.""" original = np.asarray(block, dtype=np.float64) out = original.copy() n, L = out.shape reverse = rng.random(n) < 0.06 if reverse.any(): out[reverse] = out[reverse, ::-1] if not preserve_nonnegative: invert = rng.random(n) < 0.04 if invert.any(): out[invert] *= -1.0 # goethe censor_rows = np.nonzero(rng.random(n) < 0.06)[0] for row in censor_rows: q = float(rng.uniform(0.03, 0.18)) if rng.random() < 0.5: out[row] = np.minimum(out[row], np.quantile(out[row], 1.0 - q)) else: out[row] = np.maximum(out[row], np.quantile(out[row], q)) quantized = np.nonzero(rng.random(n) < 0.07)[0] if quantized.size: x = out[quantized] lo = x.min(axis=1, keepdims=True) hi = x.max(axis=1, keepdims=True) levels = rng.integers(16, 257, size=(quantized.size, 1)) step = (hi - lo) / np.maximum(levels - 1, 1) safe_step = np.where(step < 1e-12, 1.0, step) out[quantized] = lo + np.rint((x - lo) / safe_step) * safe_step 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] degenerate = out.std(axis=1) < 1e-9 if degenerate.any(): out[degenerate] = original[degenerate] return out # goethe 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, het_frac: float = 0.35, period_p: np.ndarray | None = None, ) -> np.ndarray: t = np.arange(L, dtype=np.float64)[None, :] level = rng.normal(0.0, 1.0, size=(n, 1)) _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, period_p=period_p) 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 # goethe # goethe # goethe log_vol = np.cumsum(_sparse_jumps(rng, n, L, rate=2.0 / L, scale=0.35), axis=1) vol = np.exp(np.clip(log_vol, -2.0, 2.0)) het = (rng.random(n) < het_frac)[:, None] innov = innov * np.where(het, vol, 1.0) return series + _ar1_batch(innov, phi) def _regime_shift( rng: np.random.Generator, n: int, L: int, *, period_p: np.ndarray | None = None, ) -> np.ndarray: 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, period_p=period_p) * rng.uniform(0.0, 1.0, size=(n, 1)) 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, period_p: np.ndarray | None = None, ) -> np.ndarray: t = np.arange(L, dtype=np.float64)[None, :] _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))) amp = rng.uniform(0.1, 0.6, size=(n, 1)) seasonal_shape = _seasonal(rng, n, L, k_max=1, period_p=period_p) 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, *, period_p: np.ndarray | None = None, seasonal_frac: float = 0.25, ) -> np.ndarray: 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, :] # goethe # goethe # goethe seas = _seasonal(rng, n, L, k_max=2, period_p=period_p) amp = rng.uniform(0.05, 0.5, size=(n, 1)) mask = (rng.random(n) < seasonal_frac)[:, None] return x + drift + np.where(mask, seas * amp, 0.0) def _integrated( rng: np.random.Generator, n: int, L: int, *, period_p: np.ndarray | None = None, seasonal_frac: float = 0.30, ) -> np.ndarray: 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)) steps = rng.normal(0.0, 1.0, size=(n, L)) * sigma + drift walk = np.cumsum(steps, axis=1) walk2 = np.cumsum(walk, axis=1) o2 = order2[:, None] y = np.where(o2, walk2 / max(L, 1) ** 0.5, walk) # goethe seas = _seasonal(rng, n, L, k_max=2, period_p=period_p) amp = rng.uniform(0.1, 0.7, size=(n, 1)) mask = (rng.random(n) < seasonal_frac)[:, None] return y + np.where(mask, seas * amp, 0.0) def _threshold_ar(rng: np.random.Generator, n: int, L: int) -> np.ndarray: # goethe 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 return _threshold_ar_jit( np.ascontiguousarray(innov), np.ascontiguousarray(phi_hi), np.ascontiguousarray(phi_lo), np.ascontiguousarray(const_hi), np.ascontiguousarray(const_lo), ) def _chaotic(rng: np.random.Generator, n: int, L: int) -> np.ndarray: # goethe map_id = rng.integers(0, 3, size=n) r_log = rng.uniform(3.6, 4.0, size=n) r_sin = rng.uniform(0.85, 1.0, size=n) r_tent = rng.uniform(1.2, 1.99, size=n) r_a = np.where(map_id == 0, r_log, np.where(map_id == 1, r_sin, r_tent)) x0 = rng.uniform(0.05, 0.95, size=n) return _chaotic_jit( np.ascontiguousarray(map_id), np.ascontiguousarray(r_a), np.ascontiguousarray(x0), L, ) def _spectral_gp(rng: np.random.Generator, n: int, L: int) -> np.ndarray: """goethe""" f = _rfftfreq_cached(L)[None, :] lengthscale = np.exp(rng.uniform(np.log(8.0), np.log(256.0), size=(n, 1))) envelope = np.exp(-0.5 * (2.0 * np.pi * lengthscale * f) ** 2) z = rng.standard_normal((n, f.shape[1])) + 1j * rng.standard_normal((n, f.shape[1])) z[:, 0] = 0.0 x = np.fft.irfft(z * np.sqrt(envelope), n=L, axis=1) sd = x.std(axis=1, keepdims=True) return x / np.where(sd < 1e-12, 1.0, sd) def _long_memory(rng: np.random.Generator, n: int, L: int) -> np.ndarray: """goethe""" f = _rfftfreq_cached(L) safe_f = np.maximum(f, 1.0 / L)[None, :] beta = rng.uniform(-0.6, 2.4, size=(n, 1)) amp = safe_f ** (-0.5 * beta) multiscale = rng.random((n, 1)) < 0.4 split_idx = rng.integers(8, max(9, f.size // 3), size=(n, 1)) split_f = np.maximum(split_idx / L, 1.0 / L) beta_hi = rng.uniform(-0.6, 2.8, size=(n, 1)) above = np.arange(f.size)[None, :] > split_idx amp_hi = split_f ** (-0.5 * beta) * (safe_f / split_f) ** (-0.5 * beta_hi) amp = np.where(multiscale & above, amp_hi, amp) amp[:, 0] = 0.0 z = rng.standard_normal((n, f.size)) + 1j * rng.standard_normal((n, f.size)) x = np.fft.irfft(z * amp, n=L, axis=1) integrate = rng.random(n) < 0.25 if integrate.any(): x[integrate] = np.cumsum(x[integrate], axis=1) x -= x.mean(axis=1, keepdims=True) sd = x.std(axis=1, keepdims=True) return x / np.where(sd < 1e-12, 1.0, sd) def _ou_stochastic_vol( rng: np.random.Generator, n: int, L: int, *, period_p: np.ndarray | None = None, ) -> np.ndarray: """Regime-switching mean reversion with bounded stochastic volatility. The state recurrence now runs through the compiled ``_ar1_recur`` kernel (same math as the previous per-row ``scipy.lfilter`` 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) 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, period_p=period_p) * 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: 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) 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 x0 = mean[:, 0] + vol[:, 0] * eps[:, 0] out = _ar1_recur(drive, phi.reshape(-1), x0) 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, *, period_p: np.ndarray | None = None, ) -> np.ndarray: """Generic physical measurements without matching one private dataset.""" seasonal = _seasonal(rng, n, L, k_max=2, period_p=period_p) 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, *, period_p: np.ndarray | None = None, ) -> np.ndarray: """Seasonal Poisson/negative-binomial counts with decaying bursts.""" t = np.arange(L, dtype=np.float64)[None, :] p = _SEASONAL_PROBS if period_p is None else period_p period = _SEASONAL_PERIODS[rng.choice(_SEASONAL_PERIODS.shape[0], size=(n, 1), p=p)] 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)) ) calendar = rng.random((n, 1)) < 0.35 day_period = rng.choice(np.array([24, 48, 96, 144], dtype=np.int64), size=(n, 1)) day_idx = (np.floor_divide(np.arange(L, dtype=np.int64)[None, :], day_period) % 7) 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) # goethe # goethe # goethe 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) overdispersed = rng.random((n, 1)) < 0.5 shape = rng.uniform(0.5, 4.0, size=(n, 1)) # goethe # goethe # goethe 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: # goethe t = np.arange(L, dtype=np.float64)[None, :] base_p = rng.uniform(0.03, 0.35, size=(n, 1)) period = rng.choice([7.0, 12.0, 24.0, 48.0, 168.0], size=(n, 1)) season = rng.uniform(0.2, 1.2, size=(n, 1)) * np.sin( 2.0 * np.pi * t / period + rng.uniform(0.0, 2.0 * np.pi, size=(n, 1)) ) logit = np.log(base_p / (1.0 - base_p)) + season p = 1.0 / (1.0 + np.exp(-logit)) occur = (rng.random((n, L)) < p).astype(np.float64) magnitude = ( rng.gamma(shape=2.0, scale=1.0, size=(n, L)) * rng.uniform(1.0, 10.0, size=(n, 1)) * np.exp(0.25 * season) ) baseline = rng.uniform(0.0, 0.5, size=(n, 1)) return baseline + occur * magnitude def _pulse_outlier( rng: np.random.Generator, n: int, L: int, *, period_p: np.ndarray | None = None, ) -> np.ndarray: # goethe # goethe base = _spectral_gp(rng, n, L) * rng.uniform(0.5, 2.0, size=(n, 1)) base += _seasonal(rng, n, L, k_max=1, period_p=period_p) * rng.uniform(0.0, 1.0, size=(n, 1)) sharp = _sparse_jumps(rng, n, L, rate=3.0 / L, scale=rng.uniform(3.0, 8.0, size=n)) impulses = _sparse_jumps(rng, n, L, rate=2.0 / L, scale=rng.uniform(2.0, 7.0, size=n)) recovery = _ar1_batch(impulses, rng.uniform(0.75, 0.995, size=n)) series = base + sharp + recovery _apply_hold_runs(rng, series, rate=2.0 / L) return series # goethe def _sanitize(block: np.ndarray) -> np.ndarray: """goethe""" 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