"""zenfro_v7 — lattice prior for 4096-context → 64-step forecast skill. A deterministic mixture of dynamical stems, domain slices (retail, weather, grid load), and structure-algebra productions (layer / segment / nest / lead-lag / short-horizon / phrase-tile). Exact fractional noise and forecastable event timing live inside families rather than as extra slots. Contract: one ``np.random.default_rng(seed)``, NumPy/SciPy only, finite bounded rows via ``_sanitize``. """ 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 # Prefetch chunk size. Keeps peak memory at O(_CHUNK · max_len) under stream # feed modes that may request millions of series and stop early. _CHUNK = 2304 # Segmented AR(1) block width (~2*sqrt(L) Python iters vs scanning all L). _AR1_BLK = 40 # Multi-cadence seasonal bank 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() # Paired cadences (daily↔weekly, short↔long) drawn as coupled periods # so the cached trig bank stays the only transcendental work. Values live in # _SEASONAL_PERIODS already. _SEASONAL_PAIRS = np.array( [[15, 60], [60, 240], [24, 168], [48, 336], [96, 672], [7, 365], [12, 52], [24, 720]], dtype=np.float64, ) # ── mixture over process families ─────────────────────────────────────────── # Lattice prior: dynamics core + domain slices + structure algebra. # Exact fGn (Davies–Harte) and forecastable pulses live inside families. _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", "weather", "grid_load", "layered", "segmented", "carrier_mod", "lead_lag", "short_horizon", "phrase_tile", ) _DEFAULT_WEIGHTS: dict[str, float] = { # Dynamics core — primary 4096→64 transfer mass. "trend_seasonal_ar": 0.09, "regime_shift": 0.095, "multiplicative": 0.05, "ar2": 0.12, "integrated": 0.09, "threshold_ar": 0.03, "chaotic": 0.008, "spectral_gp": 0.045, "long_memory": 0.05, "ou_stochastic_vol": 0.075, "physical_sensors": 0.008, "seasonal_counts": 0.012, "intermittent": 0.012, "pulse_outlier": 0.015, # Domain slices (retail / weather / grid) stay minority mass. "retail_demand": 0.06, "weather": 0.03, "grid_load": 0.04, # Structure algebra — stacked / spliced / nested / lagged phrases. "layered": 0.04, "segmented": 0.025, "carrier_mod": 0.025, "lead_lag": 0.02, "short_horizon": 0.04, "phrase_tile": 0.02, } class Generator(DataGenerator): """Lattice mixture 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() # Length-normalized bimodal trend: draw total excursion, not slope*t. 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]") # Structure-algebra knobs. self._struct_depth = int(cfg.get("struct_depth", 3)) self._struct_mul_frac = float(cfg.get("struct_mul_frac", 0.35)) self._struct_cuts = int(cfg.get("struct_cuts", 2)) self._struct_nest = float(cfg.get("struct_nest", 6.0)) self._struct_lag_frac = float(cfg.get("struct_lag_frac", 0.08)) self._struct_horizon = int(cfg.get("struct_horizon", 64)) self._struct_motif_max = int(cfg.get("struct_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-v7-lattice")) def generate(self, n_series: int) -> Iterator[np.ndarray]: # Chunked lazy yield — required under streaming feed # (``corpus_mode = "stream_cpu"``): the trainer may call # ``generate(n_upper)`` where ``n_upper = token_budget // min_length + 2`` # (often millions) and stop once the token budget is spent # (cascade/trainer/stream.py). Building all ``n_series`` eagerly # would OOM before the first yield. One CHUNK at a time # keeps memory O(CHUNK) and exits early when the consumer stops, # with a fixed draw order so the sequence stays seed-deterministic. if n_series <= 0: return rng = np.random.default_rng(self._seed) max_len = self._max_len # Trend knobs are bound as builder kwargs (no module-level # mutable 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, _grid_load, partial(_layered, depth=self._struct_depth, mul_frac=self._struct_mul_frac, hi_frac=self._tr_hi_frac, exc_lo=self._tr_exc_lo, exc_hi=self._tr_exc_hi), partial(_segmented, n_cuts=self._struct_cuts, hi_frac=self._tr_hi_frac, exc_lo=self._tr_exc_lo, exc_hi=self._tr_exc_hi), partial(_carrier_mod, nest_ratio=self._struct_nest), partial(_lead_lag, lag_frac=self._struct_lag_frac), partial(_short_horizon, horizon=self._struct_horizon, 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(_phrase_tile, motif_max=self._struct_motif_max, hi_frac=self._tr_hi_frac, exc_lo=self._tr_exc_lo, exc_hi=self._tr_exc_hi), ) # One-slot prefetch: overlap NumPy/SciPy generation with training without # changing RNG ownership 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 sample a FULL _CHUNK (yield only what is still # needed) so series i stays 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] # Keep non-negativity on count/magnitude families. preserve_nonnegative = family in { "multiplicative", "physical_sensors", "seasonal_counts", "intermittent", "retail_demand", "weather", "grid_load", } 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 ) # Optionally blend a small share of full-context rows # across families (cross-family mixup). Disabled for # variable-length configs because alignment is 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 mimic late-start sensors and # left-padded histories without altering 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-v7-prefetch", 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 covers [0, _CHUNK); raise if a slot is empty. if arr is None: # pragma: no cover - defensive raise RuntimeError("internal: unfilled series slot") yield arr finally: stop.set() producer.join(timeout=1.0) # ── vectorised helpers ────────────────────────────────────────────────────── 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 params for every row (fixed RNG order), but # evaluate only active rows. Stationary terms reuse the cadence # bank via sin(a+b), skipping a fresh n×L trig pass. 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]) ) # Minority of components get slow amplitude/phase drift; most stay # stationary so clean periodic reconstruction remains learnable. modulated = np.nonzero((k > j) & (rng.random(n) < 0.35))[0] if modulated.size: # Remap global row ids 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 # Event rates are O(1/L); draw magnitudes only for hits instead of # 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. Sparse measurement effects. Thresholds use an initial observed prefix so history never depends on the unseen forecast window. ``rate_scale`` scales 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 yields an (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)) # Bimodal end-to-end excursion — length-invariant trend strength. _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) # Weekday profile on a minority of rows (calendar-shaped residual). cal = rng.random(n) < 0.28 if cal.any(): profile = rng.normal(0.0, 1.0, size=(n, 7)) profile -= profile.mean(axis=1, keepdims=True) phase = rng.integers(0, 7, size=(n, 1)) day = (np.arange(L)[None, :] + phase) % 7 weekly = np.take_along_axis(profile, day, axis=1) amp = rng.uniform(0.05, 0.45, size=(n, 1)) series = series + np.where(cal[:, None], amp * weekly, 0.0) 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 level from cumsum of sparse jumps, plus piecewise # variance regimes and light 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 beside abrupt level jumps. Sparse slope # changes create ramps/recoveries without I(2)-scale blow-up # process — covers step/ramp recoveries without I(2) blow-up. 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, :] # Bimodal log-growth excursion (same length-normalized idea as 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: # Partial autocorrelations in (-1, 1) mapped to AR(2) coeffs via # Levinson–Durbin (stationarity). Bias p1 high for # persistent / 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 integrated family without carving a separate mixture slot. """ 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 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): coeffs flip with the sign of the previous value — a simple # nonlinear recurrence with 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 maps: logistic x←r x(1-x) with r∈[3.6,4.0], and the # sine map r sin(π x). Both stay in [0,1]. A random # observation cadence 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 spectral envelope on a 2L embedding so the eval 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. Regime-switching mean reversion with clustered volatility. Regime paths, seasonal means, and heavy-tail masks are sampled in blocks; only the state recurrence scans time, vectorised across rows. """ # Toggle fast/quiet vs slow/volatile regimes. Cumulative XOR # yields persistent Markov-like paths without a per-row time 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 reversion speed per row lets SciPy run the recurrence in # compiled code. Regimes still switch equilibrium mean and vol; # rows cover fast/quiet and slow/persistent 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 for every row # would waste most 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) # Draw shock magnitudes only at the O(n) event locations. 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 get explicit calendar interaction: intraday cadence plus # seven day 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 through row-specific decay create bursts # without looping timesteps in Python. 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) # Gamma-mixed Poisson is NB marginally and gives # realistic overdispersion; half the rows stay plain 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. 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: """Weather archetypes: diurnal base, night-floored radiation, and cloud cover. Cloud red-noise is expensive, so it only runs on a minority of weather rows. Keep mixture weight low 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 _grid_load(rng: np.random.Generator, n: int, L: int) -> np.ndarray: """Non-negative electricity-style load with diurnal + weekly shape. Combines a smooth daily curve, a 7-day profile (weekend dip/lift), slow demand drift, and mild AR noise. Optional evening peak skew matches common regional load patterns without copying any private series. """ t = np.arange(L, dtype=np.float64)[None, :] tn = t / max(L - 1, 1) # Diurnal: fundamental + optional 2nd harmonic (morning/evening peaks). phase = rng.uniform(0.0, 2.0 * np.pi, size=(n, 1)) a1 = rng.uniform(0.35, 1.2, size=(n, 1)) a2 = rng.uniform(0.05, 0.45, size=(n, 1)) * (rng.random((n, 1)) < 0.7) diurnal = a1 * np.sin(2.0 * np.pi * t / 24.0 + phase) + a2 * np.sin( 4.0 * np.pi * t / 24.0 + phase + rng.uniform(-0.4, 0.4, size=(n, 1)) ) # Weekly profile with optional weekend attenuation. profile = rng.normal(0.0, 1.0, size=(n, 7)) profile -= profile.mean(axis=1, keepdims=True) weekend = rng.uniform(-1.2, 0.2, size=(n, 1)) profile[:, 5:] += weekend profile -= profile.mean(axis=1, keepdims=True) day_phase = rng.integers(0, 7, size=(n, 1)) day = (np.arange(L)[None, :] + day_phase) % 7 weekly = rng.uniform(0.08, 0.40, size=(n, 1)) * np.take_along_axis( profile, day, axis=1 ) excursion = rng.normal(0.0, 1.0, size=(n, 1)) * rng.uniform(0.2, 1.4, size=(n, 1)) latent = np.cumsum( rng.normal(0.0, 1.0, size=(n, L)) * rng.uniform(0.004, 0.03, size=(n, 1)), axis=1, ) latent = 2.5 * np.tanh(latent / 2.5) phi = rng.uniform(0.55, 0.95, size=n) sigma = rng.uniform(0.03, 0.18, size=(n, 1)) noise = _ar1_batch(rng.normal(0.0, 1.0, size=(n, L)) * sigma, phi) # Sparse spikes (outages / cold snaps) with fast recovery. impulses = _sparse_jumps( rng, n, L, rate=1.5 / L, scale=rng.uniform(0.4, 1.8, size=n) ) shocks = _ar1_batch(impulses, rng.uniform(0.80, 0.97, size=n)) base = rng.uniform(np.log(80.0), np.log(8000.0), size=(n, 1)) log_load = np.clip( base + diurnal + weekly + excursion * tn + latent + noise + shocks, np.log(1.0), np.log(5.0e5), ) return np.exp(log_load) 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 — useful for short-horizon event prediction. """ 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 # ── structure algebra ────────────────────────────────────────────────────── # Lightweight stems for stacked/spliced productions. Full builders remain # top-level families; compositions need many stems per row, so stems stay # FFT/AR/seasonal only. 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 _layered(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) rather than a single process. """ 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 _segmented(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. Each clause is an independent stem; breakpoints change the 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 _carrier_mod(rng: np.random.Generator, n: int, L: int, *, nest_ratio: float = 6.0) -> np.ndarray: """Production: Envelope ⋉ Carrier — slow scale nests a fast carrier. Smooth long-scale envelope modulates amplitude (and sometimes phase) of a faster seasonal or AR carrier — multi-scale nesting for synoptic / cycle structure. """ 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 _lead_lag(rng: np.random.Generator, n: int, L: int, *, lag_frac: float = 0.08) -> np.ndarray: """Production: Driver ▷ Response — lagged temporal causal chain. Univariate response driven by a latent driver with a drawn lag and FIR-like coupling, plus its own AR residual — teaches lead-lag structure. """ 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 _short_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. Smooth seasonal signal continuing across the short forecast window, plus an AR residual whose correlation length is O(H) so noise averages inside the 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: mix of H and H/2 correlation lengths so both half-window and # full-window noise structure appear in the prior. phi_h = float(np.exp(-1.0 / H)) phi_h2 = float(np.exp(-1.0 / max(H // 2, 8))) use_half = rng.random(n) < 0.40 phi = np.where( use_half, rng.uniform(max(0.45, phi_h2 - 0.12), min(0.97, phi_h2 + 0.08), size=n), rng.uniform(max(0.5, phi_h - 0.15), min(0.98, phi_h + 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 _phrase_tile( 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. 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 # ── finite / bounded 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