generator.py

"""zenfro_v5 — multi-source blend of the five best local priors.

Temporal grammar (informal CFG)::

    Series     → Production | Stem | Domain
    Production → Compose | Splice | Nested | Causal | Horizon | Motif
    Domain     → retail_demand | weather(light)
    Stem       → {trend_seasonal, ar2, integrated, spectral, ou, …}

Determinism, code-only, bounded+finite contracts unchanged: one
``np.random.default_rng(seed)``, allowlisted NumPy/SciPy, ``_sanitize`` gate.
"""

from __future__ import annotations

import json
from collections.abc import Iterator
from functools import lru_cache, partial
from pathlib import Path
from queue import Full, Queue
from threading import Event, Thread

import numpy as np
from scipy.signal import lfilter

from cascade.interface import DataGenerator

# Series generated per vectorised batch. Bounds peak memory to O(_CHUNK · max_len)
# so streaming feed modes (which request millions of series and stop early) never
# materialise the full corpus. Prefetching holds at most two completed chunks
# (current + queued) while the producer may build the next. The base block is
# 2048 × 4096 × 8 B = 64 MiB per base family block, plus temporary arrays.
# This remains comfortably below the 4 GiB sandbox cap. On the reference local
# A100 environment, 2048 rows generated ~6% more points/s than 1024 while 4096
# regressed slightly, so 2048 is the measured throughput sweet spot.
_CHUNK = 2560
# Block size for the segmented AR(1) scan (~2*sqrt(L) Python iters vs L).
_AR1_BLK = 32

# Multi-cadence seasonal bank (coverage) with  / 
# calendar bias: elevate 7 / 12 / 24 / 168 / 720 while retaining long-range
# periods needed for 4096-context transfer.
_SEASONAL_PERIODS = np.array(
    [4, 7, 12, 15, 24, 30, 48, 52, 60, 90, 96, 144, 168, 183, 240, 288,
     336, 365, 672, 720, 730],
    dtype=np.float64,
)
_SEASONAL_PROBS = np.array(
    [0.015, 0.12, 0.08, 0.025, 0.14, 0.025, 0.05, 0.025, 0.045, 0.015,
     0.07, 0.04, 0.12, 0.015, 0.04, 0.035, 0.025, 0.03, 0.025, 0.05, 0.015],
    dtype=np.float64,
)
_SEASONAL_PROBS /= _SEASONAL_PROBS.sum()

# Coupled calendar periods teach daily/weekly and short/long cadence
# interactions explicitly. Every value is already in _SEASONAL_PERIODS, so the
# cached sine/cosine bank remains the only trigonometric work.
_SEASONAL_PAIRS = np.array(
    [[15, 60], [60, 240], [24, 168], [48, 336], [96, 672], [7, 365],
     [12, 52], [24, 720]],
    dtype=np.float64,
)

# ── family mixture ──────────────────────────────────────────────────────────
# Dynamics-heavy v21 core + light domain families (retail, weather) + GoT.
# Weather is deliberately ~3.5% (not v16's 0.45) so seasonal/long-range mass
# is not starved on multi-domain pools (_v2 lesson).
_FAMILIES: tuple[str, ...] = (
    "trend_seasonal_ar",
    "regime_shift",
    "multiplicative",
    "ar2",
    "integrated",
    "threshold_ar",
    "chaotic",
    "spectral_gp",
    "long_memory",
    "ou_stochastic_vol",
    "physical_sensors",
    "seasonal_counts",
    "intermittent",
    "pulse_outlier",
    "retail_demand",  #  /  weekly demand
    "weather",        # archetypes, light mass
    "got_compose",
    "got_splice",
    "got_nested",
    "got_causal",
    "got_horizon",
    "got_motif",
)
_DEFAULT_WEIGHTS: dict[str, float] = {
    "trend_seasonal_ar": 0.095,
    "regime_shift": 0.09,
    "multiplicative": 0.055,
    "ar2": 0.11,
    "integrated": 0.085,
    "threshold_ar": 0.04,
    "chaotic": 0.015,
    "spectral_gp": 0.05,
    "long_memory": 0.04,
    "ou_stochastic_vol": 0.07,
    "physical_sensors": 0.01,
    "seasonal_counts": 0.015,
    "intermittent": 0.015,
    "pulse_outlier": 0.01,
    "retail_demand": 0.05,
    "weather": 0.035,
    "got_compose": 0.055,
    "got_splice": 0.04,
    "got_nested": 0.035,
    "got_causal": 0.025,
    "got_horizon": 0.035,
    "got_motif": 0.025,
}


class Generator(DataGenerator):
    """Multi-source blend prior. Submit as ``generator.Generator``."""

    def __init__(self, config_dir: str, *, seed: int) -> None:
        cfg_path = Path(config_dir) / "config.json"
        cfg = json.loads(cfg_path.read_text(encoding="utf-8")) if cfg_path.is_file() else {}
        self._cfg = cfg
        self._seed = int(seed)
        self._min_len = int(cfg.get("min_length", 64))
        self._max_len = int(cfg.get("max_length", 4096))  # = [training] context_length (train on full context)
        if self._min_len < 1 or self._max_len < self._min_len:
            raise ValueError(f"invalid length band [{self._min_len}, {self._max_len}]")
        weights = dict(_DEFAULT_WEIGHTS)
        for k, v in dict(cfg.get("family_weights", {})).items():
            if k in weights:
                weights[k] = float(v)
        w = np.asarray([weights[f] for f in _FAMILIES], dtype=np.float64)
        if not np.all(np.isfinite(w)) or w.min() < 0 or w.sum() <= 0:
            raise ValueError("family_weights must be finite, non-negative, and not all zero")
        self._weights = w / w.sum()
        # v3.9 length-NORMALIZED bimodal trend knobs (trend excursion is length-invariant;
        # real trend-strength is ~0.02 and length-invariant, but v2's slope*t grows with L).
        self._tr_hi_frac = float(cfg.get("tr_hi_frac", 0.25))
        self._tr_exc_lo = float(cfg.get("tr_exc_lo", 0.4))
        self._tr_exc_hi = float(cfg.get("tr_exc_hi", 3.0))
        self._gr_exc_lo = float(cfg.get("gr_exc_lo", 0.3))
        self._gr_exc_hi = float(cfg.get("gr_exc_hi", 2.0))
        self._sa_clean_frac = float(cfg.get("sa_clean_frac", 0.4))
        self._sa_clean_lo = float(cfg.get("sa_clean_lo", 0.02))
        self._sa_clean_hi = float(cfg.get("sa_clean_hi", 0.12))
        self._integrated_heavy_frac = float(
            cfg.get("integrated_heavy_frac", 0.25)
        )
        self._integrated_sv_frac = float(cfg.get("integrated_sv_frac", 0.30))
        self._augment = dict(cfg.get("augment", {}))
        for name, value in (
            ("integrated_heavy_frac", self._integrated_heavy_frac),
            ("integrated_sv_frac", self._integrated_sv_frac),
            ("augment.tsmixup", float(self._augment.get("tsmixup", 0.0))),
            ("augment.pad_prefix", float(self._augment.get("pad_prefix", 0.0))),
        ):
            if not 0.0 <= value <= 1.0:
                raise ValueError(f"{name} must be in [0, 1]")
        # GoT knobs — composition depth, splice density, nest/horizon/motif.
        self._got_depth = int(cfg.get("got_depth", 3))
        self._got_mul_frac = float(cfg.get("got_mul_frac", 0.35))
        self._got_splice_cuts = int(cfg.get("got_splice_cuts", 2))
        self._got_nest_ratio = float(cfg.get("got_nest_ratio", 6.0))
        self._got_causal_lag_frac = float(cfg.get("got_causal_lag_frac", 0.08))
        self._got_horizon_steps = int(cfg.get("got_horizon_steps", 64))
        self._got_motif_max = int(cfg.get("got_motif_max", 96))
        self._artifact_scale = float(cfg.get("artifact_scale", 1.0))

    @property
    def name(self) -> str:
        return str(self._cfg.get("name", "zenfro-v5-best-blend"))

    def generate(self, n_series: int) -> Iterator[np.ndarray]:
        # Lazy, chunked generation. This is REQUIRED for the streaming feed
        # modes (chain.toml ``corpus_mode = "stream_cpu"``): the trainer calls
        # ``generate(n_upper)`` with ``n_upper = token_budget // min_length + 2``
        # — often millions — and stops pulling once the token budget is hit
        # (see cascade/trainer/stream.py). Materialising all ``n_series`` up
        # front would OOM before the first yield. Generating one CHUNK at a time
        # keeps memory at O(CHUNK) and stops early when the consumer stops,
        # while a fixed draw order keeps the whole sequence seed-deterministic.
        if n_series <= 0:
            return
        rng = np.random.default_rng(self._seed)
        max_len = self._max_len
        # Bind the trend-excursion knobs as explicit builder arguments (no shared
        # module state) so the corpus is a pure function of (seed, config).
        builders = (
            partial(_trend_seasonal_ar, hi_frac=self._tr_hi_frac,
                    exc_lo=self._tr_exc_lo, exc_hi=self._tr_exc_hi,
                    clean_frac=self._sa_clean_frac,
                    clean_lo=self._sa_clean_lo, clean_hi=self._sa_clean_hi),
            _regime_shift,
            partial(_multiplicative, hi_frac=self._tr_hi_frac,
                    exc_lo=self._gr_exc_lo, exc_hi=self._gr_exc_hi),
            _ar2,
            partial(
                _integrated,
                heavy_frac=self._integrated_heavy_frac,
                sv_frac=self._integrated_sv_frac,
            ),
            _threshold_ar, _chaotic, _spectral_gp,
            _long_memory, _ou_stochastic_vol, _physical_sensors,
            _seasonal_counts, _intermittent, _pulse_outlier,
            _retail_demand, _weather,
            partial(_got_compose, depth=self._got_depth, mul_frac=self._got_mul_frac,
                    hi_frac=self._tr_hi_frac, exc_lo=self._tr_exc_lo, exc_hi=self._tr_exc_hi),
            partial(_got_splice, n_cuts=self._got_splice_cuts,
                    hi_frac=self._tr_hi_frac, exc_lo=self._tr_exc_lo, exc_hi=self._tr_exc_hi),
            partial(_got_nested, nest_ratio=self._got_nest_ratio),
            partial(_got_causal, lag_frac=self._got_causal_lag_frac),
            partial(_got_horizon, horizon=self._got_horizon_steps,
                    hi_frac=self._tr_hi_frac, exc_lo=self._tr_exc_lo, exc_hi=self._tr_exc_hi,
                    clean_frac=self._sa_clean_frac, clean_lo=self._sa_clean_lo,
                    clean_hi=self._sa_clean_hi),
            partial(_got_motif, motif_max=self._got_motif_max,
                    hi_frac=self._tr_hi_frac, exc_lo=self._tr_exc_lo, exc_hi=self._tr_exc_hi),
        )
        # Generate one chunk ahead on a CPU thread while the consumer trains on
        # the current chunk. The isolation benchmark measured 21.9% of training
        # wall blocked in next(); a one-slot queue overlaps NumPy/SciPy work
        # (which releases the GIL) without changing the RNG owner or draw order.
        queue: Queue[object] = Queue(maxsize=1)
        stop = Event()
        done = object()

        def put(item: object) -> bool:
            while not stop.is_set():
                try:
                    queue.put(item, timeout=0.1)
                    return True
                except Full:
                    continue
            return False

        def produce() -> None:
            try:
                produced = 0
                while produced < n_series and not stop.is_set():
                    # Always draw a FULL _CHUNK (yielding only what's still
                    # needed), so series i remains a pure function of (seed, i).
                    lengths = rng.integers(
                        self._min_len, max_len + 1, size=_CHUNK
                    )
                    fam_ids = rng.choice(
                        len(_FAMILIES), size=_CHUNK, p=self._weights
                    )
                    chunk: list[np.ndarray | None] = [None] * _CHUNK
                    for fam in range(len(_FAMILIES)):
                        idx = np.nonzero(fam_ids == fam)[0]
                        if idx.size == 0:
                            continue
                        block = builders[fam](rng, int(idx.size), max_len)
                        family = _FAMILIES[fam]
                        # Preserve positivity for count/magnitude families and
                        # exact integer/causal structure for count processes
                        # ( family-aware artifacts).
                        preserve_nonnegative = family in {
                            "multiplicative",
                            "physical_sensors",
                            "seasonal_counts",
                            "intermittent",
                            "retail_demand",
                            "weather",
                        }
                        if family == "retail_demand":
                            preserve_integers: bool | np.ndarray = np.all(
                                block == np.rint(block), axis=1
                            )
                        else:
                            preserve_integers = family in {
                                "seasonal_counts",
                                "intermittent",
                            }
                        # Reverse only families whose laws remain valid under
                        # time reversal. Indices: TSA, multiplicative,
                        # spectral_gp, long_memory.
                        allow_reverse = fam in (0, 2, 7, 8)
                        # Prefix-calibrated hard bounds flatten random walks;
                        # keep range artifacts off integrated paths.
                        allow_range = family != "integrated"
                        block = _sanitize(
                            _measurement_artifacts(
                                rng,
                                block,
                                preserve_nonnegative=preserve_nonnegative,
                                preserve_integers=preserve_integers,
                                allow_reverse=allow_reverse,
                                allow_range_artifacts=allow_range,
                                rate_scale=self._artifact_scale,
                            )
                        )
                        for row, series_i in enumerate(idx):
                            length = int(lengths[series_i])
                            chunk[series_i] = np.ascontiguousarray(
                                block[row, :length], dtype=np.float64
                            )

                    # Mix a conservative share of complete, full-context rows
                    # across families. This follows the useful augmentation in
                    # longrange-sv while avoiding it for variable-length
                    # configs, where alignment would be ambiguous.
                    if self._min_len == max_len:
                        mix_rate = float(self._augment.get("tsmixup", 0.0))
                        mixed = np.nonzero(rng.random(_CHUNK) < mix_rate)[0]
                        for series_i in mixed:
                            source = chunk[series_i]
                            if source is None:  # pragma: no cover - defensive
                                continue
                            n_other = int(rng.integers(1, 3))
                            others = rng.integers(0, _CHUNK, size=n_other)
                            weights = rng.dirichlet(np.ones(n_other + 1))
                            combined = weights[0] * source
                            valid = True
                            for j, other_i in enumerate(others):
                                other = chunk[int(other_i)]
                                if other is None:  # pragma: no cover - defensive
                                    valid = False
                                    break
                                combined = combined + weights[j + 1] * other
                            if valid:
                                chunk[series_i] = _sanitize(combined)

                    # Constant prefixes represent late-starting sensors and
                    # left-padded histories without changing forecast-tail
                    # dynamics.
                    pad_rate = float(self._augment.get("pad_prefix", 0.0))
                    padded = np.nonzero(rng.random(_CHUNK) < pad_rate)[0]
                    for series_i in padded:
                        series = chunk[series_i]
                        if series is None or series.size < 8:
                            continue
                        cut = int(rng.integers(series.size // 8, 3 * series.size // 4))
                        series[:cut] = series[cut]
                    take = min(_CHUNK, n_series - produced)
                    if not put((chunk, take)):
                        return
                    produced += take
            except BaseException as exc:  # propagate producer failures
                put(exc)
            finally:
                put(done)

        producer = Thread(target=produce, name="zenfro-v5-generator", daemon=True)
        producer.start()
        try:
            while True:
                item = queue.get()
                if item is done:
                    break
                if isinstance(item, BaseException):
                    raise item
                chunk, take = item
                for arr in chunk[:take]:
                    # fam_ids partitions [0, _CHUNK); fail loud if that changes.
                    if arr is None:  # pragma: no cover - defensive
                        raise RuntimeError("internal: unfilled series slot")
                    yield arr
        finally:
            stop.set()
            producer.join(timeout=1.0)


# ── shared vectorised primitives ────────────────────────────────────────────


def _ar1_batch(
    innov: np.ndarray, phi: np.ndarray, S: int = _AR1_BLK
) -> np.ndarray:
    """AR(1) filter ``x[:,t] = phi*x[:,t-1] + innov[:,t]`` along time.

    Segmented (block) scan from ``~S + L/S Python iterations
    instead of L, vectorised across the batch. Mathematically identical to the
    sequential recurrence for ``phi ∈ [0, 1)`` (FP relative error ~1e-13).
    Falls back to SciPy ``lfilter`` per row when any ``|phi| >= 1``.
    """
    n, L = innov.shape
    p = np.asarray(phi, dtype=np.float64).reshape(n)
    if np.any(np.abs(p) >= 1.0):
        x = np.empty((n, L), dtype=np.float64)
        for i in range(n):
            x[i] = lfilter([1.0], [1.0, -float(p[i])], innov[i])
        return x
    if L < 2 * S:
        x = np.empty((n, L), dtype=np.float64)
        x[:, 0] = innov[:, 0]
        for t in range(1, L):
            x[:, t] = p * x[:, t - 1] + innov[:, t]
        return x
    B = L // S
    body = B * S
    main = innov[:, :body].reshape(n, B, S)
    y = np.empty((n, B, S), dtype=np.float64)
    y[:, :, 0] = main[:, :, 0]
    pcol = p[:, None]
    for s in range(1, S):
        y[:, :, s] = pcol * y[:, :, s - 1] + main[:, :, s]
    r = p ** S
    ylast = y[:, :, S - 1]
    X = np.empty((n, B), dtype=np.float64)
    X[:, 0] = ylast[:, 0]
    for b in range(1, B):
        X[:, b] = r * X[:, b - 1] + ylast[:, b]
    carry_in = np.empty((n, B), dtype=np.float64)
    carry_in[:, 0] = 0.0
    carry_in[:, 1:] = X[:, :-1]
    ppow = p[:, None] ** np.arange(1, S + 1, dtype=np.float64)[None, :]
    x = (y + carry_in[:, :, None] * ppow[:, None, :]).reshape(n, body)
    if body == L:
        return x
    out = np.empty((n, L), dtype=np.float64)
    out[:, :body] = x
    prev = out[:, body - 1]
    for t in range(body, L):
        prev = p * prev + innov[:, t]
        out[:, t] = prev
    return out


def _ar2_batch(innov: np.ndarray, a1: np.ndarray, a2: np.ndarray) -> np.ndarray:
    """AR(2) filter: ``x_t = a1 x_{t-1} + a2 x_{t-2} + e_t`` (batched over n)."""
    n, L = innov.shape
    x = np.empty((n, L), dtype=np.float64)
    for i in range(n):
        x[i] = lfilter(
            [1.0], [1.0, -float(a1[i]), -float(a2[i])], innov[i]
        )
    return x


def _stationary_unit_ar1(
    rng: np.random.Generator, phi: np.ndarray, L: int
) -> np.ndarray:
    """Exact zero-mean, unit-variance stationary Gaussian AR(1) rows."""
    p = np.asarray(phi, dtype=np.float64).reshape(-1)
    innov = rng.standard_normal((p.size, L)) * np.sqrt(
        np.maximum(1.0 - p[:, None] * p[:, None], 1e-12)
    )
    innov[:, 0] = rng.standard_normal(p.size)
    return _ar1_batch(innov, p)


def _prefix_mean_std(
    x: np.ndarray, *, calibration_points: int = 512
) -> tuple[np.ndarray, np.ndarray]:
    """Location/scale from an initial calibration prefix only (causal)."""
    prefix = x[:, : min(x.shape[1], calibration_points)]
    mean = prefix.mean(axis=1, keepdims=True)
    std = prefix.std(axis=1, keepdims=True)
    return mean, np.where(std < 1e-12, 1.0, std)


def _prefix_standardize(
    x: np.ndarray, *, center: bool = True, calibration_points: int = 512
) -> np.ndarray:
    mean, std = _prefix_mean_std(x, calibration_points=calibration_points)
    return (x - mean) / std if center else x / std


@lru_cache(maxsize=4)
def _seasonal_basis(L: int) -> tuple[np.ndarray, np.ndarray]:
    """Cached unit sine/cosine waves for the fixed cadence bank."""
    angle = (
        2.0
        * np.pi
        * np.arange(L, dtype=np.float64)[None, :]
        / _SEASONAL_PERIODS[:, None]
    )
    return np.sin(angle), np.cos(angle)


def _seasonal(rng: np.random.Generator, n: int, L: int, k_max: int = 3) -> np.ndarray:
    """Sum of 1..k_max stationary or slowly modulated seasonal components."""
    t = np.arange(L, dtype=np.float64)[None, :]
    sin_basis, cos_basis = _seasonal_basis(L)
    k = rng.integers(1, k_max + 1, size=n)
    pair = _SEASONAL_PAIRS[
        rng.integers(0, len(_SEASONAL_PAIRS), size=n)
    ]
    use_pair = rng.random(n) < 0.35
    out = np.zeros((n, L), dtype=np.float64)
    for j in range(k_max):
        active = np.nonzero(k > j)[0]
        per = rng.choice(
            _SEASONAL_PERIODS, size=n, p=_SEASONAL_PROBS
        )
        if j < 2:
            per = np.where(use_pair, pair[:, j], per)
        per = per[:, None]
        amp = rng.uniform(0.2, 2.0, size=n)[:, None]
        phase = rng.uniform(0.0, 2.0 * np.pi, size=n)[:, None]
        # Draw parameters for every row to preserve the fixed RNG sequence, but
        # evaluate only active rows. Stationary components reuse the cadence
        # bank via sin(a+b), avoiding a fresh transcendental pass over n×L.
        basis_idx = np.searchsorted(_SEASONAL_PERIODS, per[active, 0])
        component = amp[active] * (
            sin_basis[basis_idx] * np.cos(phase[active])
            + cos_basis[basis_idx] * np.sin(phase[active])
        )
        # Real seasonal strength and timing drift. TempoPFN's strongest
        # non-SDE ablation was its complex-seasonality prior, so a minority of
        # components receive slow amplitude and phase modulation while the
        # stationary baseline remains well represented.
        modulated = np.nonzero((k > j) & (rng.random(n) < 0.35))[0]
        if modulated.size:
            # Map global row indices into the active component block.
            modulated_local = np.searchsorted(active, modulated)
            modulated_arg = (
                2.0 * np.pi * t / per[modulated] + phase[modulated]
            )
            m_per = np.clip(
                per[modulated] * rng.uniform(
                    4.0, 12.0, size=(modulated.size, 1)
                ),
                32.0,
                2.0 * L,
            )
            m_phase = rng.uniform(
                0.0, 2.0 * np.pi, size=(modulated.size, 1)
            )
            slow = np.sin(2.0 * np.pi * t / m_per + m_phase)
            amp_mod = 1.0 + rng.uniform(
                0.05, 0.45, size=(modulated.size, 1)
            ) * slow
            phase_mod = rng.uniform(
                0.05, 0.75, size=(modulated.size, 1)
            ) * np.sin(2.0 * np.pi * t / (1.7 * m_per) - m_phase)
            component[modulated_local] = (
                amp[modulated]
                * amp_mod
                * np.sin(modulated_arg + phase_mod)
            )
        out[active] += component
    return out


def _sparse_jumps(rng: np.random.Generator, n: int, L: int, rate: float, scale) -> np.ndarray:
    """A (n, L) block of mostly-zero values with occasional N(0, scale) jumps.

    ``cumsum`` over this yields a piecewise-constant level; ``exp(cumsum)`` of a
    scaled version yields a piecewise-constant positive multiplier.
    """
    mask = rng.random((n, L)) < rate
    mask[:, 0] = False
    rows, cols = np.nonzero(mask)
    jumps = np.zeros((n, L), dtype=np.float64)
    if rows.size == 0:
        return jumps
    # Rates are O(1/L), so draw magnitudes only for actual events rather than
    # allocating and filling a second dense n×L normal array.
    s = np.asarray(scale, dtype=np.float64)
    event_scale = s if s.ndim == 0 else s.reshape(n)[rows]
    jumps[rows, cols] = rng.normal(0.0, 1.0, size=rows.size) * event_scale
    return jumps


def _row_standardize(x: np.ndarray) -> np.ndarray:
    x = x - x.mean(axis=1, keepdims=True)
    sd = x.std(axis=1, keepdims=True)
    return x / np.where(sd < 1e-12, 1.0, sd)


def _measurement_artifacts(
    rng: np.random.Generator,
    block: np.ndarray,
    *,
    preserve_nonnegative: bool,
    preserve_integers: bool | np.ndarray = False,
    allow_reverse: bool = True,
    allow_range_artifacts: bool = True,
    rate_scale: float = 1.0,
) -> np.ndarray:
    """Apply sparse, cheap real-measurement effects to a generated block.

    TempoPFN reports a 5.4% aggregate CRPS gain from its complete augmentation
    pipeline. Rates are conservative. Sensor thresholds are calibrated from an
    initial observed prefix () so history does not depend on the
    unseen forecast target. ``rate_scale`` multiplies base rates.
    """
    original = np.asarray(block, dtype=np.float64)
    out = original.copy()
    n, L = out.shape
    rs = float(np.clip(rate_scale, 0.0, 3.0))
    calibration_len = min(L, 512)

    reverse = (rng.random(n) < (0.06 * rs)) if allow_reverse else np.zeros(n, dtype=bool)
    out[reverse] = out[reverse, ::-1]

    if not preserve_nonnegative:
        invert = rng.random(n) < (0.04 * rs)
        out[invert] *= -1.0

    # Draw selections even when range artifacts are disabled so RNG sequences
    # stay comparable across family-aware flags.
    for row in np.nonzero(rng.random(n) < (0.06 * rs))[0]:
        q = float(rng.uniform(0.03, 0.18))
        upper = rng.random() < 0.5
        if not allow_range_artifacts:
            continue
        calibration = out[row, :calibration_len]
        if upper:
            threshold = np.quantile(calibration, 1.0 - q)
            out[row] = np.minimum(out[row], threshold)
        else:
            threshold = np.quantile(calibration, q)
            out[row] = np.maximum(out[row], threshold)

    quantized = np.nonzero(rng.random(n) < (0.07 * rs))[0]
    if quantized.size:
        levels = rng.integers(16, 257, size=(quantized.size, 1))
        if allow_range_artifacts:
            x = out[quantized]
            calibration = x[:, :calibration_len]
            lo = calibration.min(axis=1, keepdims=True)
            hi = calibration.max(axis=1, keepdims=True)
            step = (hi - lo) / np.maximum(levels - 1, 1)
            safe_step = np.where(step < 1e-12, 1.0, step)
            clipped = np.clip(x, lo, hi)
            out[quantized] = (
                lo + np.rint((clipped - lo) / safe_step) * safe_step
            )

    held = np.nonzero(rng.random(n) < (0.04 * rs))[0]
    if held.size:
        factors = rng.choice([2, 4, 8], size=held.size, p=[0.55, 0.30, 0.15])
        for factor in (2, 4, 8):
            rows = held[factors == factor]
            if rows.size:
                out[rows] = np.repeat(
                    out[rows, ::factor], factor, axis=1
                )[:, :L]

    if isinstance(preserve_integers, np.ndarray):
        integer_rows = np.asarray(preserve_integers, dtype=bool).reshape(n)
        out[integer_rows] = np.maximum(np.rint(out[integer_rows]), 0.0)
    elif preserve_integers:
        out = np.maximum(np.rint(out), 0.0)

    degenerate = out[:, :calibration_len].std(axis=1) < 1e-9
    out[degenerate] = original[degenerate]
    return out


# ── family builders: each returns a (n, L) float64 block ────────────────────


def _trend_seasonal_ar(rng: np.random.Generator, n: int, L: int, *,
                       hi_frac: float = 0.25, exc_lo: float = 0.4,
                       exc_hi: float = 3.0, clean_frac: float = 0.4,
                       clean_lo: float = 0.02, clean_hi: float = 0.12) -> np.ndarray:
    t = np.arange(L, dtype=np.float64)[None, :]
    level = rng.normal(0.0, 1.0, size=(n, 1))
    # v3: bimodal trend. The total trend EXCURSION over the series is drawn directly
    # (0..exc across t/(L-1)), so the trend sits ~16x below v2's slope*t — v2's linear
    # trend was a measured ~16x too strong vs real data at production lengths.
    _hi = rng.random((n, 1)) < hi_frac
    exc = np.where(_hi, rng.normal(0.0, exc_hi, size=(n, 1)),
                   rng.normal(0.0, exc_lo, size=(n, 1)))
    tn = t / max(L - 1, 1)
    series = level + exc * tn + _seasonal(rng, n, L)
    phi = rng.uniform(0.0, 0.85, size=n)
    clean = rng.random((n, 1)) < clean_frac
    sigma = np.where(
        clean,
        rng.uniform(clean_lo, clean_hi, size=(n, 1)),
        rng.uniform(0.1, 0.6, size=(n, 1)),
    )
    innov = rng.normal(0.0, 1.0, size=(n, L)) * sigma
    return series + _ar1_batch(innov, phi)


def _regime_shift(rng: np.random.Generator, n: int, L: int) -> np.ndarray:
    # Piecewise-constant level via cumsum of sparse jumps, plus a piecewise
    # variance regime (occasional volatility multiplier), plus mild seasonality.
    level = np.cumsum(_sparse_jumps(rng, n, L, rate=3.0 / L, scale=2.0), axis=1)
    log_vol = np.cumsum(_sparse_jumps(rng, n, L, rate=3.0 / L, scale=0.5), axis=1)
    vol = np.exp(np.clip(log_vol, -3.0, 3.0)) * rng.uniform(0.1, 0.5, size=(n, 1))
    noise = rng.normal(0.0, 1.0, size=(n, L)) * vol
    seas = _seasonal(rng, n, L, k_max=2) * rng.uniform(0.0, 1.0, size=(n, 1))
    # Piecewise-affine drift complements abrupt level jumps. Sparse slope
    # changes create ramps and recoveries without the explosive scale of an I(2)
    # process, covering TempoPFN's high-impact Step/Sawtooth structures.
    slope = rng.normal(0.0, 1.0 / L, size=(n, 1)) + np.cumsum(
        _sparse_jumps(rng, n, L, rate=2.0 / L, scale=4.0 / L), axis=1
    )
    piecewise_trend = np.cumsum(slope, axis=1)
    return level + piecewise_trend + seas + noise


def _multiplicative(rng: np.random.Generator, n: int, L: int, *,
                    hi_frac: float = 0.25, exc_lo: float = 0.3, exc_hi: float = 2.0) -> np.ndarray:
    t = np.arange(L, dtype=np.float64)[None, :]
    # v3: bimodal log-growth excursion (drawn directly), same rationale as the linear trend.
    _hg = rng.random((n, 1)) < hi_frac
    gexc = np.where(_hg, rng.normal(0.0, exc_hi, size=(n, 1)),
                    rng.normal(0.0, exc_lo, size=(n, 1)))
    tn = t / max(L - 1, 1)
    base_level = np.exp(gexc * tn + rng.normal(0.0, 0.3, size=(n, 1)))  # positive, drifting
    amp = rng.uniform(0.1, 0.6, size=(n, 1))
    seasonal_shape = _seasonal(rng, n, L, k_max=1)
    seasonal_sd = seasonal_shape.std(axis=1, keepdims=True)
    seasonal_shape /= np.where(seasonal_sd < 1e-12, 1.0, seasonal_sd)
    seas = 1.0 + amp * seasonal_shape
    noise = 1.0 + rng.normal(0.0, 1.0, size=(n, L)) * rng.uniform(0.02, 0.15, size=(n, 1))
    scale = rng.uniform(1.0, 50.0, size=(n, 1))
    return scale * base_level * np.clip(seas, 0.05, None) * np.clip(noise, 0.05, None)


def _ar2(rng: np.random.Generator, n: int, L: int) -> np.ndarray:
    # Draw partial autocorrelations in (-1, 1) and map to AR(2) coeffs via
    # Levinson-Durbin, which guarantees stationarity. Bias p1 high for
    # persistent (sometimes near-unit-root) series.
    p1 = rng.uniform(0.3, 0.98, size=n)
    p2 = rng.uniform(-0.6, 0.6, size=n)
    a2 = p2
    a1 = p1 * (1.0 - p2)
    sigma = rng.uniform(0.2, 0.8, size=(n, 1))
    innov = rng.normal(0.0, 1.0, size=(n, L)) * sigma
    x = _ar2_batch(innov, a1, a2)
    drift = rng.normal(0.0, 0.005, size=(n, 1)) * np.arange(L, dtype=np.float64)[None, :]
    return x + drift


def _integrated(
    rng: np.random.Generator,
    n: int,
    L: int,
    *,
    heavy_frac: float = 0.25,
    sv_frac: float = 0.30,
) -> np.ndarray:
    """I(1)/I(2) paths with selective heavy tails and clustered volatility.

    The Gaussian baseline remains the majority. Heavy rows use variance-scaled
    Student-t innovations, while stochastic-volatility rows receive a smooth
    AR(1) log-vol multiplier. These mechanisms are applied inside an existing
     family rather than funding a new family at the expense of its
    measured mixture.
    """
    order2 = rng.random(n) < 0.35
    drift = rng.normal(0.0, 0.02, size=(n, 1))
    sigma = rng.uniform(0.2, 1.0, size=(n, 1))
    eps = rng.normal(0.0, 1.0, size=(n, L))
    heavy = np.nonzero(rng.random(n) < heavy_frac)[0]
    if heavy.size:
        df = rng.uniform(3.0, 12.0, size=(heavy.size, 1))
        eps[heavy] = rng.standard_t(df, size=(heavy.size, L)) / np.sqrt(
            df / (df - 2.0)
        )

    stochastic = np.nonzero(rng.random(n) < sv_frac)[0]
    if stochastic.size:
        phi = 0.995
        log_vol = lfilter(
            [1.0],
            [1.0, -phi],
            rng.standard_normal((stochastic.size, L)),
            axis=1,
        )
        log_vol -= log_vol.mean(axis=1, keepdims=True)
        log_vol /= np.maximum(log_vol.std(axis=1, keepdims=True), 1e-9)
        log_vol *= rng.uniform(0.10, 0.55, size=(stochastic.size, 1))
        eps[stochastic] *= np.exp(np.clip(log_vol, -2.0, 2.0))

    steps = eps * sigma + drift
    walk = np.cumsum(steps, axis=1)
    walk2 = np.cumsum(walk, axis=1)
    o2 = order2[:, None]
    # I(2) grows fast; damp it so it shares scale with the I(1) branch.
    return np.where(o2, walk2 / max(L, 1) ** 0.5, walk)


def _threshold_ar(rng: np.random.Generator, n: int, L: int) -> np.ndarray:
    # SETAR(2): coefficient flips with the sign of the previous value — a simple
    # nonlinear recurrence that produces asymmetric, regime-switching dynamics.
    phi_hi = rng.uniform(0.3, 0.9, size=n)
    phi_lo = rng.uniform(-0.9, 0.3, size=n)
    const_hi = rng.normal(0.0, 0.3, size=n)
    const_lo = rng.normal(0.0, 0.3, size=n)
    sigma = rng.uniform(0.2, 0.7, size=(n, 1))
    innov = rng.normal(0.0, 1.0, size=(n, L)) * sigma
    x = np.empty((n, L), dtype=np.float64)
    x[:, 0] = innov[:, 0]
    for t in range(1, L):
        prev = x[:, t - 1]
        hi = prev >= 0.0
        phi = np.where(hi, phi_hi, phi_lo)
        const = np.where(hi, const_hi, const_lo)
        x[:, t] = np.clip(const + phi * prev + innov[:, t], -1e6, 1e6)
    return x


def _chaotic(rng: np.random.Generator, n: int, L: int) -> np.ndarray:
    # Bounded chaotic maps: logistic x_{t+1}=r x(1-x) with r∈[3.6,4.0], and the
    # sine map r sin(pi x). Both stay in [0,1]; standardise afterwards. A random
    # observation length as a "sampling rate" adds variety across series.
    use_sine = rng.random(n) < 0.5
    r_log = rng.uniform(3.6, 4.0, size=n)
    r_sin = rng.uniform(0.85, 1.0, size=n)
    x0 = rng.uniform(0.05, 0.95, size=n)
    x = np.empty((n, L), dtype=np.float64)
    cur = x0.copy()
    x[:, 0] = cur
    for t in range(1, L):
        nxt_log = r_log * cur * (1.0 - cur)
        nxt_sin = r_sin * np.sin(np.pi * cur)
        cur = np.where(use_sine, nxt_sin, nxt_log)
        cur = np.clip(cur, 0.0, 1.0)
        x[:, t] = cur
    return x


def _spectral_gp(rng: np.random.Generator, n: int, L: int) -> np.ndarray:
    """Smooth stationary GP-like paths sampled in O(n L log L).

    An RBF kernel has a Gaussian spectral density. Drawing complex Fourier
    coefficients under that envelope and applying one batched inverse FFT
    preserves the useful smoothness/length-scale prior without the old
    48-pass cosine loop.
    """
    f = np.fft.rfftfreq(L)[None, :]
    lengthscale = np.exp(rng.uniform(np.log(8.0), np.log(256.0), size=(n, 1)))
    envelope = np.exp(-0.5 * (2.0 * np.pi * lengthscale * f) ** 2)
    z = rng.standard_normal((n, f.shape[1])) + 1j * rng.standard_normal((n, f.shape[1]))
    z[:, 0] = 0.0
    x = np.fft.irfft(z * np.sqrt(envelope), n=L, axis=1)
    sd = x.std(axis=1, keepdims=True)
    return x / np.where(sd < 1e-12, 1.0, sd)


def _long_memory(rng: np.random.Generator, n: int, L: int) -> np.ndarray:
    """Fractional power-law paths with both persistent and rough regimes.

    The spectral slope beta spans anti-persistent noise through persistent
    long-memory levels. A minority of rows are integrated once to include
    nonstationary fBm-like paths; row standardisation keeps scales bounded.
    """
    f = np.fft.rfftfreq(L)
    safe_f = np.maximum(f, 1.0 / L)[None, :]
    beta = rng.uniform(-0.6, 2.4, size=(n, 1))
    amp = safe_f ** (-0.5 * beta)
    # Some rows change roughness above a random frequency, giving smooth
    # large-scale structure and rough local variation (or the reverse) without
    # another FFT. Match amplitudes at the split to avoid a spectral jump.
    multiscale = rng.random((n, 1)) < 0.4
    split_idx = rng.integers(8, max(9, f.size // 3), size=(n, 1))
    split_f = np.maximum(split_idx / L, 1.0 / L)
    beta_hi = rng.uniform(-0.6, 2.8, size=(n, 1))
    above = np.arange(f.size)[None, :] > split_idx
    amp_hi = split_f ** (-0.5 * beta) \
        * (safe_f / split_f) ** (-0.5 * beta_hi)
    amp = np.where(multiscale & above, amp_hi, amp)
    amp[:, 0] = 0.0
    z = rng.standard_normal((n, f.size)) + 1j * rng.standard_normal((n, f.size))
    x = np.fft.irfft(z * amp, n=L, axis=1)
    integrate = rng.random(n) < 0.25
    if integrate.any():
        x[integrate] = np.cumsum(x[integrate], axis=1)
    x -= x.mean(axis=1, keepdims=True)
    sd = x.std(axis=1, keepdims=True)
    return x / np.where(sd < 1e-12, 1.0, sd)


def _ou_stochastic_vol(rng: np.random.Generator, n: int, L: int) -> np.ndarray:
    """Regime-switching mean reversion with bounded stochastic volatility.

    This is a CPU-cheap discrete Euler/AR analogue of TempoPFN's highest-impact
    OU SDE prior. Regime paths, seasonal means, volatility envelopes, and
    heavy-tail masks are sampled in whole blocks; only the state recurrence
    scans time, vectorised across all rows.
    """
    # Toggle between a fast/quiet and a slow/volatile regime. A cumulative XOR
    # builds persistent Markov-like paths without a per-row Python loop.
    switch_rate = np.exp(rng.uniform(np.log(0.001), np.log(0.15), size=(n, 1)))
    switches = rng.random((n, L)) < switch_rate
    switches[:, 0] = rng.random(n) < 0.5
    regime = np.bitwise_and(np.cumsum(switches, axis=1), 1).astype(np.int8)

    # One mean-reversion speed per row lets SciPy execute the recurrence in
    # compiled code. Regime paths still switch equilibrium mean and volatility;
    # rows span both fast/quiet and slow/persistent reversion rates.
    slow = rng.random((n, 1)) < 0.5
    phi = np.where(
        slow,
        rng.uniform(0.995, 0.9995, size=(n, 1)),
        rng.uniform(0.90, 0.99, size=(n, 1)),
    )
    mu0 = rng.normal(-2.0, 1.0, size=(n, 1))
    mu1 = rng.normal(2.0, 1.0, size=(n, 1))
    mean = np.where(regime == 0, mu0, mu1)
    seasonal_on = rng.random((n, 1)) < 0.6
    mean += seasonal_on * _seasonal(rng, n, L, k_max=3) \
        * rng.uniform(0.5, 3.0, size=(n, 1))

    sigma0 = rng.lognormal(np.log(0.3), 0.3, size=(n, 1))
    sigma1 = rng.lognormal(np.log(1.5), 0.5, size=(n, 1))
    base_sigma = np.where(regime == 0, sigma0, sigma1)
    log_vol = np.cumsum(
        _sparse_jumps(rng, n, L, rate=8.0 / L, scale=0.35), axis=1
    )
    log_vol -= log_vol.mean(axis=1, keepdims=True)
    vol = base_sigma * np.exp(np.clip(log_vol, -1.5, 1.5))

    eps = rng.standard_normal((n, L))
    heavy = np.nonzero(rng.random(n) < 0.35)[0]
    if heavy.size:
        # Replace only heavy-tailed rows; drawing Student-t noise for every row
        # previously discarded 65% of that relatively expensive work.
        eps[heavy] = (
            rng.standard_t(4.0, size=(heavy.size, L)) / np.sqrt(2.0)
        )
    shocks = rng.random((n, L)) < (3.0 / L)
    shock_rows, shock_cols = np.nonzero(shocks)
    # As with sparse jumps, draw shock magnitudes only at the O(n) events.
    eps[shock_rows, shock_cols] += rng.normal(
        0.0, 5.0, size=shock_rows.size
    )

    innovation_scale = np.sqrt(np.maximum(1.0 - phi * phi, 1e-6))
    drive = (1.0 - phi) * mean + innovation_scale * vol * eps
    out = np.empty((n, L), dtype=np.float64)
    out[:, 0] = mean[:, 0] + vol[:, 0] * eps[:, 0]
    for i in range(n):
        p = float(phi[i, 0])
        out[i, 1:] = lfilter(
            [1.0], [1.0, -p], drive[i, 1:], zi=[p * out[i, 0]]
        )[0]

    scale = np.exp(rng.uniform(np.log(0.1), np.log(50.0), size=(n, 1)))
    shift = rng.uniform(-100.0, 100.0, size=(n, 1))
    return out * scale + shift


def _physical_sensors(rng: np.random.Generator, n: int, L: int) -> np.ndarray:
    """Generic physical measurements without matching one private dataset.

    Four row-level archetypes cover smooth signed measurements, bounded
    percentages, pressure-like wandering levels, and non-negative skewed
    magnitudes. All share multi-cadence seasonality, smooth synoptic variation,
    and sparse fronts/gusts.
    """
    seasonal = _seasonal(rng, n, L, k_max=2)
    smooth = _spectral_gp(rng, n, L)
    fronts = np.cumsum(
        _sparse_jumps(rng, n, L, rate=5.0 / L, scale=1.0), axis=1
    )
    base = (
        seasonal * rng.uniform(0.3, 2.0, size=(n, 1))
        + smooth * rng.uniform(0.2, 1.2, size=(n, 1))
        + fronts * rng.uniform(0.2, 1.0, size=(n, 1))
    )

    kind = rng.integers(0, 4, size=n)
    out = base.copy()

    bounded = kind == 1
    if bounded.any():
        gain = rng.uniform(0.8, 3.5, size=(int(bounded.sum()), 1))
        midpoint = rng.uniform(-0.8, 0.8, size=(int(bounded.sum()), 1))
        out[bounded] = 100.0 / (1.0 + np.exp(-gain * (base[bounded] - midpoint)))

    pressure = kind == 2
    if pressure.any():
        count = int(pressure.sum())
        walk = np.cumsum(rng.standard_normal((count, L)), axis=1) / np.sqrt(L)
        level = rng.uniform(900.0, 1100.0, size=(count, 1))
        out[pressure] = level + rng.uniform(2.0, 15.0, size=(count, 1)) * walk \
            + 2.0 * fronts[pressure] + 0.5 * seasonal[pressure]

    magnitude = kind == 3
    if magnitude.any():
        count = int(magnitude.sum())
        gusts = (rng.random((count, L)) < (8.0 / L)) \
            * rng.lognormal(0.0, 0.8, size=(count, L))
        power = rng.uniform(1.0, 1.6, size=(count, 1))
        out[magnitude] = np.abs(base[magnitude]) ** power + gusts

    return out


def _seasonal_counts(rng: np.random.Generator, n: int, L: int) -> np.ndarray:
    """Seasonal Poisson/negative-binomial counts with decaying bursts.

    This keeps count positivity and discreteness intact while covering
    overdispersion, cadence-linked rate variation, slow signed growth, and
    release/news-like bursts. Computation remains batched across rows.
    """
    t = np.arange(L, dtype=np.float64)[None, :]
    period = rng.choice(
        _SEASONAL_PERIODS, size=(n, 1), p=_SEASONAL_PROBS
    )
    phase = rng.uniform(0.0, 2.0 * np.pi, size=(n, 1))
    amp = rng.uniform(0.15, 0.8, size=(n, 1))
    log_rate = amp * np.sin(2.0 * np.pi * t / period + phase)
    second = rng.random((n, 1)) < 0.55
    log_rate += second * (0.5 * amp) * np.sin(
        4.0 * np.pi * t / period + rng.uniform(0.0, 2.0 * np.pi, size=(n, 1))
    )
    # A minority carry explicit calendar interaction: intraday cadence plus
    # seven day-specific factors, with a randomized weekend dip or lift.
    calendar = rng.random((n, 1)) < 0.35
    day_period = rng.choice([24, 48, 96, 144], size=(n, 1))
    day_idx = (np.floor_divide(np.arange(L)[None, :], day_period) % 7).astype(np.int64)
    day_factors = rng.normal(0.0, 0.12, size=(n, 7))
    day_factors[:, 5:] += rng.uniform(-0.8, 0.3, size=(n, 1))
    calendar_effect = np.take_along_axis(day_factors, day_idx, axis=1)
    log_rate += calendar * calendar_effect
    excursion = rng.uniform(-0.5, 0.5, size=(n, 1))
    log_rate += excursion * t / max(L - 1, 1)

    # Sparse positive impulses filtered by row-specific decay create bursts
    # without a Python loop over timesteps.
    impulses = (
        (rng.random((n, L)) < (2.0 / L))
        * rng.uniform(1.0, 10.0, size=(n, L))
    )
    burst = _ar1_batch(impulses, rng.uniform(0.85, 0.995, size=(n, 1)))
    base = np.exp(rng.uniform(np.log(3.0), np.log(3000.0), size=(n, 1)))
    lam = base * np.exp(np.clip(log_rate, -5.0, 5.0)) * (1.0 + burst)
    np.clip(lam, 0.0, 1.0e7, out=lam)

    # A gamma-mixed Poisson is negative-binomial marginally and provides
    # realistic overdispersion. Half the rows remain ordinary Poisson.
    overdispersed = rng.random((n, 1)) < 0.5
    shape = rng.uniform(0.5, 4.0, size=(n, 1))
    mixed = lam * rng.gamma(shape, 1.0 / shape, size=(n, L))
    return rng.poisson(np.where(overdispersed, mixed, lam)).astype(np.float64)


def _intermittent(rng: np.random.Generator, n: int, L: int) -> np.ndarray:
    # Seasonal zero-inflated demand with weekly occurrence bias ()
    # and a persistent latent occurrence/size state ().
    t = np.arange(L, dtype=np.float64)[None, :]
    base_p = rng.uniform(0.03, 0.35, size=(n, 1))
    period = rng.choice([7.0, 12.0, 24.0, 48.0, 168.0], size=(n, 1))
    season = rng.uniform(0.2, 1.2, size=(n, 1)) * np.sin(
        2.0 * np.pi * t / period + rng.uniform(0.0, 2.0 * np.pi, size=(n, 1))
    )
    # Mild dedicated weekly modulation for retail-like domains.
    week = 0.55 + 0.45 * (
        0.5 + 0.5 * np.sin(
            2.0 * np.pi * t / 7.0
            + rng.uniform(0.0, 2.0 * np.pi, size=(n, 1))
        )
    )
    occurrence_state = _stationary_unit_ar1(
        rng, rng.uniform(0.0, 0.95, size=n), L
    )
    occurrence_state = _prefix_standardize(occurrence_state)
    logit = (
        np.log(base_p / (1.0 - base_p))
        + season
        + np.log(np.clip(week, 0.2, 1.5))
        + rng.uniform(0.0, 1.2, size=(n, 1)) * occurrence_state
    )
    p = 1.0 / (1.0 + np.exp(-logit))
    occur = (rng.random((n, L)) < p).astype(np.float64)

    independent_size_state = _stationary_unit_ar1(
        rng, rng.uniform(0.5, 0.98, size=n), L
    )
    independent_size_state = _prefix_standardize(independent_size_state)
    coupling = rng.uniform(0.15, 0.55, size=(n, 1))
    size_state = (
        coupling * occurrence_state
        + np.sqrt(1.0 - coupling * coupling) * independent_size_state
    )
    size_eta = rng.uniform(0.15, 0.45, size=(n, 1))
    size_factor = np.exp(size_eta * size_state - 0.5 * size_eta * size_eta)
    magnitude = np.maximum(1.0, np.rint(
        rng.gamma(shape=2.0, scale=1.0, size=(n, L))
        * rng.uniform(1.0, 10.0, size=(n, 1))
        * np.exp(0.25 * season)
        * size_factor
    ))
    return occur * magnitude


def _retail_demand(rng: np.random.Generator, n: int, L: int) -> np.ndarray:
    """Weekly retail demand with persistent level drift and asymmetric events.

    From  (qweqwe/dasadas demand lineage used by _v2). Each
    row has a seven-day profile, optional adjacent-day suppression, a slow
    latent log-demand walk, and short-lived promotions or stockouts. Mixes
    integer counts with continuous non-negative magnitude series.
    """
    t = np.arange(L, dtype=np.float64)[None, :]
    tn = t / max(L - 1, 1)

    profile = rng.normal(0.0, 1.0, size=(n, 7))
    profile -= profile.mean(axis=1, keepdims=True)
    special_days = rng.random(n) < 0.55
    first_day = rng.integers(0, 7, size=n)
    depth = rng.uniform(0.4, 1.6, size=n)
    adjacent = np.zeros((n, 7), dtype=np.float64)
    rows = np.arange(n)
    adjacent[rows, first_day] -= depth
    adjacent[rows, (first_day + 1) % 7] -= depth
    adjacent -= adjacent.mean(axis=1, keepdims=True)
    profile += np.where(special_days[:, None], adjacent, 0.0)
    profile -= profile.mean(axis=1, keepdims=True)

    phase = rng.integers(0, 7, size=(n, 1))
    day_index = (np.arange(L)[None, :] + phase) % 7
    weekly = (
        rng.uniform(0.03, 0.50, size=(n, 1))
        * np.take_along_axis(profile, day_index, axis=1)
    )

    excursion = (
        rng.normal(0.0, 1.0, size=(n, 1))
        * rng.uniform(0.3, 2.5, size=(n, 1))
    )
    latent_raw = np.cumsum(
        rng.normal(0.0, 1.0, size=(n, L))
        * rng.uniform(0.005, 0.05, size=(n, 1)),
        axis=1,
    )
    latent_walk = 3.0 * np.tanh(latent_raw / 3.0)

    promotions = np.zeros((n, L), dtype=np.float64)
    promo_rows, promo_cols = np.nonzero(
        rng.random((n, L)) < (rng.uniform(1.0, 8.0, size=(n, 1)) / L)
    )
    promotions[promo_rows, promo_cols] = (
        np.abs(rng.normal(0.0, 1.0, size=promo_rows.size))
        * rng.uniform(0.5, 2.5, size=n)[promo_rows]
    )
    promo_echo = np.zeros_like(promotions)
    promo_echo[:, 1:] = (
        promotions[:, :-1] * rng.uniform(0.2, 0.6, size=(n, 1))
    )

    stockouts = np.zeros((n, L), dtype=np.float64)
    stock_rows, stock_cols = np.nonzero(
        rng.random((n, L)) < (rng.uniform(0.0, 4.0, size=(n, 1)) / L)
    )
    stockouts[stock_rows, stock_cols] = (
        np.abs(rng.normal(0.0, 1.0, size=stock_rows.size))
        * rng.uniform(0.3, 1.5, size=n)[stock_rows]
    )

    noise = (
        rng.normal(0.0, 1.0, size=(n, L))
        * rng.uniform(0.02, 0.25, size=(n, 1))
    )
    base_log_level = rng.uniform(np.log(0.2), np.log(3000.0), size=(n, 1))
    log_mean = np.clip(
        base_log_level
        + excursion * tn
        + latent_walk
        + weekly
        + promotions
        + promo_echo
        - stockouts
        + noise,
        -8.0,
        13.0,
    )
    level = np.exp(log_mean)

    count_rows = rng.random(n) < 0.35
    target_mean = np.exp(
        rng.uniform(np.log(1.0), np.log(80.0), size=(n, 1))
    )
    calibration_mean, _ = _prefix_mean_std(level)
    count_scale = target_mean / np.clip(calibration_mean, 1e-9, None)
    counts = rng.poisson(
        np.clip(level * count_scale, 0.0, 1.0e6)
    ).astype(np.float64)
    return np.where(count_rows[:, None], counts, level)


def _cloud(rng: np.random.Generator, n: int, L: int) -> np.ndarray:
    """U-shaped [0, 100] cloud fraction via multi-timescale red noise + clip."""
    t = np.arange(L, dtype=np.float64)[None, :]
    phi_s = rng.uniform(0.990, 0.9990, size=n)
    syn = _ar1_batch(rng.normal(0.0, 1.0, (n, L)), phi_s)
    syn = (syn - syn.mean(1, keepdims=True)) / (syn.std(1, keepdims=True) + 1e-9)
    phi_m = rng.uniform(0.895, 0.95, size=n)
    mid = _ar1_batch(rng.normal(0.0, 1.0, (n, L)), phi_m)
    mid = (mid - mid.mean(1, keepdims=True)) / (mid.std(1, keepdims=True) + 1e-9)
    phi_f = rng.uniform(0.75, 0.87, size=n)
    fast = _ar1_batch(rng.normal(0.0, 1.0, (n, L)), phi_f)
    fast = (fast - fast.mean(1, keepdims=True)) / (fast.std(1, keepdims=True) + 1e-9)
    g = rng.normal(0.0, 1.0, (n, L))
    spike = (rng.random((n, L)) < 0.05) * rng.uniform(5.0, 11.0, (n, L))
    heavy = g * (1.0 + spike)
    rough = heavy - 0.30 * np.concatenate([np.zeros((n, 1)), heavy[:, :-1]], axis=1)
    rough = rough / (rough.std(1, keepdims=True) + 1e-9)
    diur = np.sin(2.0 * np.pi * t / 24.0 + rng.uniform(0.0, 2.0 * np.pi, (n, 1)))
    ws = 0.95 * rng.uniform(0.8, 1.2, (n, 1))
    wm = 0.82 * rng.uniform(0.7, 1.3, (n, 1))
    wf = 0.55 * rng.uniform(0.7, 1.3, (n, 1))
    wr = 0.30 * rng.uniform(0.7, 1.3, (n, 1))
    wd = 0.44 * rng.uniform(0.3, 1.4, (n, 1))
    u = ws * syn + wm * mid + wf * fast + wr * rough + wd * diur
    centre = np.clip(rng.normal(61.0, 29.0, (n, 1)), 2.0, 98.0)
    span = 57.0 * rng.uniform(0.82, 1.18, (n, 1))
    return np.round(np.clip(centre + span * u, 0.0, 100.0))


def _weather(rng: np.random.Generator, n: int, L: int) -> np.ndarray:
    """Sliced weather archetypes: diurnal base / night-floored radiation / cloud.

    From ``ith ```` quieter AR and optional monthly
    harmonic. Assigned first so expensive cloud red-noise runs only on ~18% of
    weather rows. Keep mixture weight low (~3.5%) for multi-domain pools.
    """
    t = np.arange(L, dtype=np.float64)[None, :]
    rad_mask = rng.random(n) < 0.25
    cloud_mask = (rng.random(n) < 0.24) & ~rad_mask
    wbp_mask = ~cloud_mask
    out = np.empty((n, L), dtype=np.float64)

    wbp_idx = np.nonzero(wbp_mask)[0]
    m = wbp_idx.size
    if m:
        radb = rad_mask[wbp_idx][:, None]
        level = rng.normal(0.0, 1.0, size=(m, 1))
        a1 = rng.uniform(0.55, 2.1, size=(m, 1))
        p1 = rng.uniform(0.0, 2.0 * np.pi, size=(m, 1))
        a2 = rng.uniform(0.12, 0.65, size=(m, 1))
        p2 = rng.uniform(0.0, 2.0 * np.pi, size=(m, 1))
        seas = (
            a1 * np.sin(2.0 * np.pi * t / 24.0 + p1)
            + a2 * np.sin(2.0 * np.pi * t / 12.0 + p2)
        )
        wk = (rng.random((m, 1)) < 0.45).astype(np.float64)
        seas += wk * rng.uniform(0.12, 0.55, size=(m, 1)) * np.sin(
            2.0 * np.pi * t / 168.0
            + rng.uniform(0.0, 2.0 * np.pi, size=(m, 1))
        )
        month = (rng.random((m, 1)) < 0.35).astype(np.float64)
        seas += month * rng.uniform(0.08, 0.40, size=(m, 1)) * np.sin(
            2.0 * np.pi * t / 720.0
            + rng.uniform(0.0, 2.0 * np.pi, size=(m, 1))
        )
        yamp = rng.uniform(0.25, 1.6, size=(m, 1))
        yper = rng.uniform(2000.0, 9000.0, size=(m, 1))
        yearly = yamp * np.sin(
            2.0 * np.pi * t / yper
            + rng.uniform(0.0, 2.0 * np.pi, size=(m, 1))
        )
        phi = rng.uniform(0.65, 0.96, size=m)
        sigma = rng.uniform(0.04, 0.18, size=(m, 1))
        noise = _ar1_batch(rng.normal(0.0, 1.0, size=(m, L)) * sigma, phi)
        base = level + seas + yearly + noise
        thr = rng.uniform(0.2, 0.6, size=(m, 1)) * a1
        ramp = rng.uniform(0.8, 1.6, size=(m, 1))
        rad_diurnal = ramp * np.maximum(
            a1 * np.sin(2.0 * np.pi * t / 24.0 + p1) - thr, 0.0
        )
        night = rad_diurnal <= 0.0
        rad_series = (
            level + rad_diurnal + yearly
            + np.where(night, noise * 0.12, noise)
        )
        out[wbp_idx] = np.where(radb, rad_series, base)

    cloud_idx = np.nonzero(cloud_mask)[0]
    if cloud_idx.size:
        out[cloud_idx] = _cloud(rng, int(cloud_idx.size), L)
    return out


def _pulse_outlier(rng: np.random.Generator, n: int, L: int) -> np.ndarray:
    # A smooth base with isolated outliers, persistent shock/recovery responses,
    # and genuine held-constant runs.
    base = _spectral_gp(rng, n, L) * rng.uniform(0.5, 2.0, size=(n, 1))
    base += _seasonal(rng, n, L, k_max=1) * rng.uniform(0.0, 1.0, size=(n, 1))
    sharp = _sparse_jumps(
        rng, n, L, rate=3.0 / L, scale=rng.uniform(3.0, 8.0, size=n)
    )
    impulses = _sparse_jumps(
        rng, n, L, rate=2.0 / L, scale=rng.uniform(2.0, 7.0, size=n)
    )
    recovery = _ar1_batch(impulses, rng.uniform(0.75, 0.995, size=n))
    series = base + sharp + recovery

    # Sparse event loops, not a time-axis scan: typically two starts per row.
    starts = rng.random((n, L)) < (2.0 / L)
    starts[:, 0] = False
    for row in range(n):
        for start in np.nonzero(starts[row])[0]:
            run = int(rng.integers(3, 65))
            end = min(int(start) + run, L)
            series[row, start:end] = series[row, start - 1]
    return series


# ── Grammar-of-Time productions ─────────────────────────────────────────────
# Stem pool for compositions. Order is fixed so RNG draw sequences stay stable
# across config-only weight changes to non-GoT families.


# Lightweight stem pool for GoT productions. Full  builders stay as
# top-level families; compositions need many stems per row, so these stay
# FFT/AR/seasonal only — no Python-over-t recurrences.
def _stem_ar_seasonal(rng: np.random.Generator, n: int, L: int) -> np.ndarray:
    seas = _seasonal(rng, n, L, k_max=2)
    phi = rng.uniform(0.1, 0.9, size=n)
    sigma = rng.uniform(0.15, 0.7, size=(n, 1))
    innov = rng.normal(0.0, 1.0, size=(n, L)) * sigma
    return _row_standardize(seas * rng.uniform(0.3, 1.2, size=(n, 1)) + _ar1_batch(innov, phi))


def _stem_integrated(rng: np.random.Generator, n: int, L: int) -> np.ndarray:
    sigma = rng.uniform(0.2, 1.0, size=(n, 1))
    walk = np.cumsum(rng.normal(0.0, 1.0, size=(n, L)) * sigma, axis=1)
    return _row_standardize(walk)


def _sample_stem(rng: np.random.Generator, n: int, L: int) -> np.ndarray:
    """Draw one cheap stem family per row and fill a (n, L) block."""
    builders = (
        _stem_ar_seasonal,
        _spectral_gp,
        _long_memory,
        _stem_integrated,
        _ar2,
    )
    fam = rng.integers(0, len(builders), size=n)
    out = np.empty((n, L), dtype=np.float64)
    for k, builder in enumerate(builders):
        idx = np.nonzero(fam == k)[0]
        if idx.size == 0:
            continue
        out[idx] = _row_standardize(builder(rng, int(idx.size), L))
    return out


def _got_compose(rng: np.random.Generator, n: int, L: int, *,
                 depth: int = 3, mul_frac: float = 0.35,
                 hi_frac: float = 0.25, exc_lo: float = 0.4,
                 exc_hi: float = 2.5) -> np.ndarray:
    """Production: Stem ⊕ Stem [⊕ Stem] — additive or multiplicative phrase.

    Each row stacks ``depth`` standardised stems. A minority use multiplicative
    agreement (level × seasonal-like factor), matching TempoPFN-style compound
    structure rather than a single process family.
    """
    depth = int(np.clip(depth, 2, 4))
    # Always draw ``depth`` stems so the RNG stream is depth-stable.
    stems = [_sample_stem(rng, n, L) for _ in range(depth)]
    n_active = rng.integers(2, depth + 1, size=n)

    out = np.zeros((n, L), dtype=np.float64)
    use_mul = rng.random(n) < mul_frac
    for j, stem in enumerate(stems):
        active = (n_active > j)[:, None]
        w = rng.uniform(0.4, 1.6, size=(n, 1))
        # Additive branch.
        add_mask = active & (~use_mul[:, None])
        out = np.where(add_mask, out + w * stem, out)
        # Multiplicative branch: first stem is the carrier; later stems modulate.
        if j == 0:
            out = np.where(use_mul[:, None], stem, out)
        else:
            factor = 1.0 + 0.35 * w * stem
            out = np.where(active & use_mul[:, None], out * factor, out)

    # Mild length-normalised trend affix so compose rows still carry forecastable
    # drift without exploding scale.
    t = np.arange(L, dtype=np.float64)[None, :] / max(L - 1, 1)
    _hi = rng.random((n, 1)) < hi_frac
    exc = np.where(_hi, rng.normal(0.0, exc_hi * 0.5, size=(n, 1)),
                   rng.normal(0.0, exc_lo * 0.5, size=(n, 1)))
    out = out + exc * t

    # Sparse punctuation affix (jumps) on a minority of rows.
    punct = rng.random(n) < 0.4
    if punct.any():
        jumps = np.cumsum(
            _sparse_jumps(rng, n, L, rate=2.5 / L, scale=1.5), axis=1
        )
        out[punct] = out[punct] + jumps[punct]

    scale = np.exp(rng.uniform(np.log(0.2), np.log(40.0), size=(n, 1)))
    shift = rng.uniform(-50.0, 50.0, size=(n, 1))
    return out * scale + shift


def _got_splice(rng: np.random.Generator, n: int, L: int, *,
                n_cuts: int = 2, hi_frac: float = 0.25,
                exc_lo: float = 0.4, exc_hi: float = 2.5) -> np.ndarray:
    """Production: Stem ‖ Stem — clause boundaries splice different dynamics.

    Unlike 's piecewise level jumps inside one process, each clause is
    an independent stem; breakpoints teach structural change of *generating law*.
    """
    n_cuts = int(np.clip(n_cuts, 1, 4))
    n_clauses = n_cuts + 1
    clauses = [_sample_stem(rng, n, L) for _ in range(n_clauses)]

    # Cut positions in (0.15L, 0.85L), sorted per row.
    cuts = np.sort(
        rng.integers(max(1, L // 8), max(2, (7 * L) // 8), size=(n, n_cuts)),
        axis=1,
    )
    # Enforce strictly increasing cuts with a small gap.
    for c in range(1, n_cuts):
        cuts[:, c] = np.maximum(cuts[:, c], cuts[:, c - 1] + max(8, L // 32))
    cuts = np.clip(cuts, 1, L - 2)

    out = clauses[0].copy()
    t_idx = np.arange(L)[None, :]
    for c in range(n_cuts):
        after = t_idx >= cuts[:, c:c + 1]
        out = np.where(after, clauses[c + 1], out)

    # Soft blend near each cut so the splice is a transition, not a hard glitch
    # (real regime changes often ramp over a few steps).
    blend_w = max(4, L // 128)
    for c in range(n_cuts):
        cut = cuts[:, c:c + 1]
        dist = (t_idx - cut).astype(np.float64)
        gate = np.clip(0.5 + dist / (2.0 * blend_w), 0.0, 1.0)
        left = clauses[c]
        right = clauses[c + 1]
        near = np.abs(dist) <= blend_w
        blended = (1.0 - gate) * left + gate * right
        out = np.where(near, blended, out)

    # Optional level offset between clauses (structural break magnitude).
    level_jump = rng.normal(0.0, 1.5, size=(n, n_cuts))
    for c in range(n_cuts):
        after = t_idx >= cuts[:, c:c + 1]
        out = np.where(after, out + level_jump[:, c:c + 1], out)

    t = np.arange(L, dtype=np.float64)[None, :] / max(L - 1, 1)
    _hi = rng.random((n, 1)) < hi_frac
    exc = np.where(_hi, rng.normal(0.0, exc_hi * 0.4, size=(n, 1)),
                   rng.normal(0.0, exc_lo * 0.4, size=(n, 1)))
    out = out + exc * t

    scale = np.exp(rng.uniform(np.log(0.2), np.log(40.0), size=(n, 1)))
    shift = rng.uniform(-50.0, 50.0, size=(n, 1))
    return out * scale + shift


def _got_nested(rng: np.random.Generator, n: int, L: int, *,
                nest_ratio: float = 6.0) -> np.ndarray:
    """Production: Envelope ⋉ Carrier — slow scale nests a fast carrier.

    Hierarchical seasonality / synoptic weather / business-cycle nesting: a
    smooth long-scale envelope modulates amplitude (and sometimes phase) of a
    faster seasonal or AR carrier. This is the multi-scale grammar Chronos-style
    priors emphasise but  only touches via modulated seasonality.
    """
    nest_ratio = float(np.clip(nest_ratio, 2.0, 24.0))
    t = np.arange(L, dtype=np.float64)[None, :]

    # Slow envelope: spectral GP with long lengthscale, or long sinusoid.
    use_gp_env = rng.random(n) < 0.55
    env = np.empty((n, L), dtype=np.float64)
    gp_rows = np.nonzero(use_gp_env)[0]
    sin_rows = np.nonzero(~use_gp_env)[0]
    if gp_rows.size:
        # Force long lengthscales for the envelope.
        f = np.fft.rfftfreq(L)[None, :]
        lengthscale = np.exp(
            rng.uniform(np.log(64.0), np.log(min(512.0, L / 2.0)), size=(gp_rows.size, 1))
        )
        envelope = np.exp(-0.5 * (2.0 * np.pi * lengthscale * f) ** 2)
        z = rng.standard_normal((gp_rows.size, f.shape[1])) + 1j * rng.standard_normal(
            (gp_rows.size, f.shape[1])
        )
        z[:, 0] = 0.0
        g = np.fft.irfft(z * np.sqrt(envelope), n=L, axis=1)
        env[gp_rows] = _row_standardize(g)
    if sin_rows.size:
        per = rng.uniform(L / nest_ratio, L / 1.5, size=(sin_rows.size, 1))
        phase = rng.uniform(0.0, 2.0 * np.pi, size=(sin_rows.size, 1))
        env[sin_rows] = np.sin(2.0 * np.pi * t / per + phase)

    # Fast carrier: seasonal bank and/or AR(2).
    carrier = _seasonal(rng, n, L, k_max=3)
    carrier = _row_standardize(carrier)
    mix_ar = rng.random(n) < 0.45
    if mix_ar.any():
        ar = _row_standardize(_ar2(rng, n, L))
        w = rng.uniform(0.3, 0.7, size=(n, 1))
        carrier = np.where(mix_ar[:, None], w * carrier + (1.0 - w) * ar, carrier)

    amp = 1.0 + rng.uniform(0.3, 1.4, size=(n, 1)) * env
    # Phase wobble as a small quadrature mix with a lagged carrier — vectorised,
    # no per-row roll. Equivalent spirit: slow envelope nudges fast phase.
    carrier_lag = np.empty_like(carrier)
    carrier_lag[:, 0] = carrier[:, 0]
    carrier_lag[:, 1:] = carrier[:, :-1]
    wobble = rng.uniform(0.0, 0.35, size=(n, 1)) * env
    out = amp * (carrier + wobble * carrier_lag)

    # Residual noise scaled by envelope intensity (prosody).
    sigma = rng.uniform(0.05, 0.35, size=(n, 1)) * (0.5 + 0.5 * np.abs(env))
    phi = rng.uniform(0.0, 0.8, size=n)
    innov = rng.normal(0.0, 1.0, size=(n, L)) * sigma
    out = out + _ar1_batch(innov, phi)

    scale = np.exp(rng.uniform(np.log(0.2), np.log(40.0), size=(n, 1)))
    shift = rng.uniform(-50.0, 50.0, size=(n, 1))
    return out * scale + shift


def _delay_batch(x: np.ndarray, lags: np.ndarray) -> np.ndarray:
    """Causal delay with edge hold, vectorised over a small lag vocabulary.

    Rows sharing a lag are shifted in one slice copy — O(#unique_lags) passes
    instead of a Python loop over n.
    """
    n, L = x.shape
    out = np.empty_like(x)
    # Hold initial value for the lag prefix.
    for lag in np.unique(lags):
        rows = np.nonzero(lags == lag)[0]
        if rows.size == 0:
            continue
        lag_i = int(lag)
        block = x[rows]
        delayed = np.empty_like(block)
        delayed[:, :lag_i] = block[:, :1]
        delayed[:, lag_i:] = block[:, :-lag_i]
        out[rows] = delayed
    return out


def _got_causal(rng: np.random.Generator, n: int, L: int, *,
                lag_frac: float = 0.08) -> np.ndarray:
    """Production: Driver ▷ Response — lagged temporal causal chain.

    Inspired by Chronos-2 / CauKer temporal causal graphs, specialised to a
    univariate observable: the emitted series is a response driven by a latent
    driver with a drawn lag and FIR-like coupling, plus its own AR residual.
    Teaches lead-lag structure that pure mixture families never emit.
    """
    lag_frac = float(np.clip(lag_frac, 0.01, 0.25))
    # Discrete lag menu keeps _delay_batch on a handful of unique values.
    lag_menu = np.unique(
        np.clip(
            (np.array([0.01, 0.02, 0.04, 0.06, 0.08, 0.12, 0.16, 0.20]) * L).astype(np.int64),
            1,
            max(1, int(L * lag_frac)),
        )
    )
    lags = rng.choice(lag_menu, size=n)

    driver = _sample_stem(rng, n, L)
    use_parent2 = rng.random(n) < 0.4
    parent2 = _sample_stem(rng, n, L)

    a0 = rng.uniform(0.2, 1.2, size=(n, 1))
    a1 = rng.uniform(0.3, 1.5, size=(n, 1))
    b = rng.uniform(0.2, 1.0, size=(n, 1))
    lags2 = rng.choice(lag_menu, size=n)

    delayed = _delay_batch(driver, lags)
    resp = a0 * driver + a1 * delayed
    if use_parent2.any():
        delayed2 = _delay_batch(parent2, lags2)
        resp = np.where(use_parent2[:, None], resp + b * delayed2, resp)

    phi = rng.uniform(0.2, 0.9, size=n)
    sigma = rng.uniform(0.1, 0.5, size=(n, 1))
    innov = rng.normal(0.0, 1.0, size=(n, L)) * sigma
    resp = resp + _ar1_batch(innov, phi)
    seas_on = rng.random((n, 1)) < 0.5
    resp = resp + seas_on * _seasonal(rng, n, L, k_max=2) * rng.uniform(
        0.1, 0.8, size=(n, 1)
    )

    scale = np.exp(rng.uniform(np.log(0.2), np.log(40.0), size=(n, 1)))
    shift = rng.uniform(-50.0, 50.0, size=(n, 1))
    return resp * scale + shift


def _got_horizon(
    rng: np.random.Generator,
    n: int,
    L: int,
    *,
    horizon: int = 64,
    hi_frac: float = 0.25,
    exc_lo: float = 0.4,
    exc_hi: float = 2.5,
    clean_frac: float = 0.4,
    clean_lo: float = 0.02,
    clean_hi: float = 0.12,
) -> np.ndarray:
    """Production: Signal + short residual tuned to the eval forecast horizon.

    Cascade scores 4096-context → 64-step forecasts. This production makes that
    geometry explicit: a smooth, seasonally coherent signal that continues
    across the horizon, plus an AR residual whose correlation length is O(H)
    so noise averages inside the forecast window without erasing continuity.
    """
    H = int(np.clip(horizon, 16, 256))
    t = np.arange(L, dtype=np.float64)[None, :] / max(L - 1, 1)

    # Persistent multi-cadence signal (the forecastable backbone).
    signal = _seasonal(rng, n, L, k_max=3)
    # Mild spectral envelope so the signal is not pure sinusoids.
    mix_gp = rng.random(n) < 0.45
    if mix_gp.any():
        gp = _row_standardize(_spectral_gp(rng, n, L))
        w = rng.uniform(0.2, 0.55, size=(n, 1))
        signal = np.where(mix_gp[:, None], (1.0 - w) * signal + w * gp, signal)
    signal = _row_standardize(signal)

    _hi = rng.random((n, 1)) < hi_frac
    exc = np.where(
        _hi,
        rng.normal(0.0, exc_hi, size=(n, 1)),
        rng.normal(0.0, exc_lo, size=(n, 1)),
    )
    signal = signal + exc * t

    # Residual with phi ~ exp(-1/H) so autocorr at lag H is ~e^{-1}.
    # Clean rows shrink residual further (sharp periodic reconstruction).
    phi_target = float(np.exp(-1.0 / H))
    phi = rng.uniform(max(0.5, phi_target - 0.15), min(0.98, phi_target + 0.08), size=n)
    clean = rng.random((n, 1)) < clean_frac
    sigma = np.where(
        clean,
        rng.uniform(clean_lo, clean_hi, size=(n, 1)),
        rng.uniform(0.12, 0.55, size=(n, 1)),
    )
    innov = rng.normal(0.0, 1.0, size=(n, L)) * sigma
    residual = _ar1_batch(innov, phi)

    # Sparse punctuation that recovers inside ~H steps (event + decay).
    impulses = _sparse_jumps(rng, n, L, rate=1.5 / L, scale=rng.uniform(1.0, 4.0, size=n))
    recover_phi = rng.uniform(0.85, 0.98, size=n)
    events = _ar1_batch(impulses, recover_phi)
    use_events = rng.random(n) < 0.35
    residual = residual + use_events[:, None] * events

    out = signal + residual
    scale = np.exp(rng.uniform(np.log(0.2), np.log(40.0), size=(n, 1)))
    shift = rng.uniform(-50.0, 50.0, size=(n, 1))
    return out * scale + shift


def _got_motif(
    rng: np.random.Generator,
    n: int,
    L: int,
    *,
    motif_max: int = 96,
    hi_frac: float = 0.25,
    exc_lo: float = 0.4,
    exc_hi: float = 2.5,
) -> np.ndarray:
    """Production: Tile(local shape) — non-sinusoidal repeating phrases.

    Pure Fourier seasonality under-covers weekday/shift/ops motifs that are
    shaped bumps, not sinusoids. Each row draws a short motif, tiles it across
    L, and applies slow amplitude/level drift so consecutive periods remain
    forecastable while still evolving.
    """
    motif_max = int(np.clip(motif_max, 16, 256))
    # Prefer periods near common cadences and the 64-step forecast window.
    period_menu = np.array(
        [7, 12, 16, 24, 32, 48, 64, 72, 96], dtype=np.int64
    )
    period_menu = period_menu[period_menu <= motif_max]
    periods = rng.choice(period_menu, size=n)

    t = np.arange(L, dtype=np.float64)[None, :]
    out = np.empty((n, L), dtype=np.float64)
    for p in np.unique(periods):
        rows = np.nonzero(periods == p)[0]
        m = int(rows.size)
        p_i = int(p)
        # Shape family: raised-cosine bump, asymmetric triangle, or AR snippet.
        kind = rng.integers(0, 3, size=m)
        motif = np.empty((m, p_i), dtype=np.float64)
        u = np.linspace(0.0, 1.0, p_i, endpoint=False)[None, :]

        cos_rows = kind == 0
        if cos_rows.any():
            width = rng.uniform(0.15, 0.55, size=(int(cos_rows.sum()), 1))
            centre = rng.uniform(0.2, 0.8, size=(int(cos_rows.sum()), 1))
            motif[cos_rows] = np.maximum(
                0.0, np.cos(np.pi * (u - centre) / np.maximum(width, 1e-3))
            )

        tri_rows = kind == 1
        if tri_rows.any():
            peak = rng.uniform(0.2, 0.8, size=(int(tri_rows.sum()), 1))
            left = np.clip(u / np.maximum(peak, 1e-3), 0.0, 1.0)
            right = np.clip((1.0 - u) / np.maximum(1.0 - peak, 1e-3), 0.0, 1.0)
            motif[tri_rows] = np.minimum(left, right)

        ar_rows = kind == 2
        if ar_rows.any():
            count = int(ar_rows.sum())
            phi = rng.uniform(0.3, 0.9, size=count)
            innov = rng.normal(0.0, 1.0, size=(count, p_i))
            motif[ar_rows] = _ar1_batch(innov, phi)

        motif = _row_standardize(motif)
        # Tile
        reps = int(np.ceil(L / p_i))
        tiled = np.tile(motif, (1, reps))[:, :L]
        # Slow amplitude and level drift across tiles (forecastable evolution).
        n_tiles = max(1, int(np.ceil(L / p_i)))
        amp_path = np.cumsum(
            rng.normal(0.0, 0.08, size=(m, n_tiles)), axis=1
        )
        amp_path = 1.0 + 0.35 * _row_standardize(amp_path)
        level_path = np.cumsum(
            rng.normal(0.0, 0.05, size=(m, n_tiles)), axis=1
        )
        tile_idx = np.minimum(np.arange(L) // p_i, n_tiles - 1)
        amp = amp_path[:, tile_idx]
        level = level_path[:, tile_idx]
        # Within-period jitter so exact copies are rare.
        jitter = rng.normal(0.0, 0.05, size=(m, L))
        out[rows] = amp * tiled + level + jitter

    _hi = rng.random((n, 1)) < hi_frac
    exc = np.where(
        _hi,
        rng.normal(0.0, exc_hi * 0.5, size=(n, 1)),
        rng.normal(0.0, exc_lo * 0.5, size=(n, 1)),
    )
    tn = np.arange(L, dtype=np.float64)[None, :] / max(L - 1, 1)
    out = out + exc * tn

    # Light AR noise on top.
    phi = rng.uniform(0.0, 0.7, size=n)
    sigma = rng.uniform(0.05, 0.35, size=(n, 1))
    out = out + _ar1_batch(rng.normal(0.0, 1.0, size=(n, L)) * sigma, phi)

    scale = np.exp(rng.uniform(np.log(0.2), np.log(40.0), size=(n, 1)))
    shift = rng.uniform(-50.0, 50.0, size=(n, 1))
    return out * scale + shift




# ── final safety gate ───────────────────────────────────────────────────────


def _sanitize(block: np.ndarray) -> np.ndarray:
    """Guarantee finite float64 values and proportionally bound each row.

    The trainer's ``check_series`` rejects any non-finite value, which would
    fail the whole run. Proportional rescaling preserves within-row geometry;
    hard clipping can create artificial constant plateaus on explosive paths.
    """
    x = np.asarray(block, dtype=np.float64)
    np.nan_to_num(x, copy=False, nan=0.0, posinf=1e6, neginf=-1e6)
    if x.ndim == 1:
        peak = float(np.max(np.abs(x)))
        if peak > 1e6:
            x *= 1e6 / peak
    else:
        peak = np.max(np.abs(x), axis=1, keepdims=True)
        scale = np.where(peak > 1e6, 1e6 / np.maximum(peak, 1e-12), 1.0)
        x *= scale
    return x