generator.py

"""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