generator.py

"""zenfro_v2 — weather-tuned mixture + Grammar of Time.

Cascade trains Toto2 from scratch on this corpus alone and scores held-out
windows of length 64 after a 4096-step context. The prior must therefore teach
*forecastable* multi-scale structure under a fixed compute wall.

Evolution from zenfro_v1
-----------------------
v1's weather-heavy blend (diurnal / radiation / cloud) remains the skill
backbone. v2 adds:

  * Grammar of Time (GoT) productions — compose / splice / nested / causal —
    so Toto2 sees temporal *phrases*, not only mutually exclusive families
    (Chronos-2 / TempoPFN / CauKer style compositionality).
  * Coupled calendar seasonality (24↔168, 7↔52, …) and slow amp/phase drift.
  * Clean low-noise seasonal mode for sharp periodic reconstruction.
  * Slightly quieter innovations + retained numba kernels for wall throughput.

Hot recurrences stay numba-jitted (cache=False for cascade's dynamic loader).
Chunked emission + fixed L=4096 keep generation wall-safe under the token wall.
"""

from __future__ import annotations

import json
from collections.abc import Callable, Iterator
from pathlib import Path
from typing import Any

import numpy as np
from numba import njit

from cascade.interface import DataGenerator

_BATCH = 512
_AR1_BLOCK = 32

_PRIOR_NAMES: tuple[str, ...] = (
    "trend_seasonal_ar",
    "regime_shift",
    "multiplicative",
    "ar2",
    "integrated",
    "threshold_ar",
    "chaotic",
    "rff_gp",
    "intermittent",
    "pulse_outlier",
    "weather",
    # Grammar-of-Time productions
    "got_compose",
    "got_splice",
    "got_nested",
    "got_causal",
)

# Weather remains the largest single stem; ~22% mass on GoT so combination
# structure is learned without drowning the proven weather signal.
_PRIOR_MASS: dict[str, float] = {
    "weather": 0.34,
    "trend_seasonal_ar": 0.11,
    "regime_shift": 0.06,
    "ar2": 0.08,
    "integrated": 0.08,
    "rff_gp": 0.05,
    "multiplicative": 0.04,
    "threshold_ar": 0.015,
    "chaotic": 0.01,
    "intermittent": 0.012,
    "pulse_outlier": 0.003,
    "got_compose": 0.08,
    "got_splice": 0.06,
    "got_nested": 0.05,
    "got_causal": 0.03,
}

# Broader calendar bank than v1; bias toward diurnal / weekly / monthly.
_SEASON_PERIODS = np.asarray(
    [
        4.0,
        7.0,
        12.0,
        15.0,
        24.0,
        30.0,
        48.0,
        52.0,
        60.0,
        96.0,
        144.0,
        168.0,
        240.0,
        336.0,
        365.0,
        504.0,
        672.0,
        720.0,
    ],
    dtype=np.float64,
)
_SEASON_P = np.asarray(
    [
        0.02,
        0.11,
        0.06,
        0.03,
        0.18,
        0.04,
        0.05,
        0.04,
        0.04,
        0.05,
        0.04,
        0.14,
        0.03,
        0.04,
        0.03,
        0.03,
        0.04,
        0.07,
    ],
    dtype=np.float64,
)
_SEASON_P = _SEASON_P / _SEASON_P.sum()

# Coupled calendar pairs teach daily↔weekly and short↔long interactions.
_SEASON_PAIRS = np.asarray(
    [
        [24.0, 168.0],
        [7.0, 365.0],
        [12.0, 52.0],
        [48.0, 336.0],
        [96.0, 672.0],
        [15.0, 60.0],
        [24.0, 720.0],
    ],
    dtype=np.float64,
)


# ── numba kernels (cache=False: cascade imports this file as a dynamic module) ─


@njit(cache=False)
def _jit_ar2(innov: np.ndarray, a1: np.ndarray, a2: np.ndarray) -> np.ndarray:
    n, L = innov.shape
    out = np.empty((n, L), dtype=np.float64)
    for i in range(n):
        aa1, aa2 = a1[i], a2[i]
        out[i, 0] = innov[i, 0]
        if L > 1:
            out[i, 1] = aa1 * out[i, 0] + innov[i, 1]
        for t in range(2, L):
            out[i, t] = aa1 * out[i, t - 1] + aa2 * out[i, t - 2] + innov[i, t]
    return out


@njit(cache=False)
def _jit_setar(
    innov: np.ndarray,
    phi_pos: np.ndarray,
    phi_neg: np.ndarray,
    c_pos: np.ndarray,
    c_neg: np.ndarray,
) -> np.ndarray:
    n, L = innov.shape
    out = np.empty((n, L), dtype=np.float64)
    for i in range(n):
        out[i, 0] = innov[i, 0]
        for t in range(1, L):
            prev = out[i, t - 1]
            if prev >= 0.0:
                v = c_pos[i] + phi_pos[i] * prev + innov[i, t]
            else:
                v = c_neg[i] + phi_neg[i] * prev + innov[i, t]
            if v > 1e6:
                v = 1e6
            elif v < -1e6:
                v = -1e6
            out[i, t] = v
    return out


@njit(cache=False)
def _jit_chaos(
    pick_sine: np.ndarray,
    r_log: np.ndarray,
    r_sin: np.ndarray,
    x0: np.ndarray,
    L: int,
) -> np.ndarray:
    n = x0.shape[0]
    out = np.empty((n, L), dtype=np.float64)
    for i in range(n):
        cur = x0[i]
        out[i, 0] = cur
        sine = pick_sine[i]
        rl, rs = r_log[i], r_sin[i]
        for t in range(1, L):
            if sine:
                cur = rs * np.sin(np.pi * cur)
            else:
                cur = rl * cur * (1.0 - cur)
            if cur < 0.0:
                cur = 0.0
            elif cur > 1.0:
                cur = 1.0
            out[i, t] = cur
    return out


@njit(cache=False)
def _jit_holds(series: np.ndarray, hold: np.ndarray) -> None:
    n, L = series.shape
    for i in range(n):
        for t in range(1, L):
            if hold[i, t]:
                series[i, t] = series[i, t - 1]


# ── thin numpy wrappers ─────────────────────────────────────────────────────


def _ar1(innov: np.ndarray, phi: np.ndarray, S: int = _AR1_BLOCK) -> np.ndarray:
    """AR(1) via segmented (block) scan — ~2*sqrt(L) Python iters, same law as loop."""
    n, L = innov.shape
    p = np.asarray(phi, dtype=np.float64).reshape(n)
    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, :]
    x = x.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(innov: np.ndarray, a1: np.ndarray, a2: np.ndarray) -> np.ndarray:
    return _jit_ar2(
        np.ascontiguousarray(innov),
        np.ascontiguousarray(a1.reshape(-1)),
        np.ascontiguousarray(a2.reshape(-1)),
    )


def _pick_periods(rng: np.random.Generator, n: int) -> np.ndarray:
    return rng.choice(_SEASON_PERIODS, size=n, p=_SEASON_P)[:, None]


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


def _harmonics(
    rng: np.random.Generator,
    n: int,
    L: int,
    *,
    depth: int = 3,
    couple_frac: float = 0.35,
    modulate_frac: float = 0.30,
) -> np.ndarray:
    """Sum of up to ``depth`` sinusoids with optional calendar coupling + drift."""
    t = np.arange(L, dtype=np.float64)[None, :]
    n_comp = rng.integers(1, depth + 1, size=n)
    pair = _SEASON_PAIRS[rng.integers(0, len(_SEASON_PAIRS), size=n)]
    use_pair = rng.random(n) < couple_frac
    acc = np.zeros((n, L), dtype=np.float64)
    for j in range(depth):
        on = (n_comp > j).astype(np.float64)[:, None]
        period = _pick_periods(rng, n)
        if j < 2:
            period = np.where(use_pair[:, None], pair[:, j : j + 1], period)
        lo, hi = (0.45, 2.3) if j == 0 else (0.15, 1.15)
        amp = rng.uniform(lo, hi, size=(n, 1))
        phase = rng.uniform(0.0, 2.0 * np.pi, size=(n, 1))
        arg = 2.0 * np.pi * t / period + phase
        # Minority: slow amplitude / phase modulation (TempoPFN-style complex seas).
        mod_on = (rng.random(n) < modulate_frac) & (n_comp > j)
        if mod_on.any():
            m_per = np.clip(
                period * rng.uniform(4.0, 12.0, size=(n, 1)),
                32.0,
                2.0 * L,
            )
            m_phase = rng.uniform(0.0, 2.0 * np.pi, size=(n, 1))
            slow = np.sin(2.0 * np.pi * t / m_per + m_phase)
            amp_mod = 1.0 + rng.uniform(0.05, 0.40, size=(n, 1)) * slow
            phase_mod = rng.uniform(0.05, 0.70, size=(n, 1)) * np.sin(
                2.0 * np.pi * t / (1.7 * m_per) - m_phase
            )
            wave = amp * amp_mod * np.sin(arg + phase_mod)
            base = amp * np.sin(arg)
            wave = np.where(mod_on[:, None], wave, base)
        else:
            wave = amp * np.sin(arg)
        acc += on * wave
    return acc


def _jumps(rng: np.random.Generator, n: int, L: int, rate: float, scale) -> np.ndarray:
    hit = rng.random((n, L)) < rate
    mag = rng.normal(0.0, 1.0, size=(n, L))
    s = np.asarray(scale, dtype=np.float64)
    if s.ndim == 1:
        s = s[:, None]
    out = hit * mag * s
    out[:, 0] = 0.0
    return out


def _finite(block: np.ndarray) -> np.ndarray:
    x = np.asarray(block, dtype=np.float64)
    x = np.nan_to_num(x, nan=0.0, posinf=1e6, neginf=-1e6)
    return np.clip(x, -1e6, 1e6)


def _rescale(x: np.ndarray, rng: np.random.Generator) -> np.ndarray:
    scale = np.exp(rng.uniform(np.log(0.2), np.log(40.0), size=(x.shape[0], 1)))
    shift = rng.uniform(-50.0, 50.0, size=(x.shape[0], 1))
    return x * scale + shift


# ── family emitters ─────────────────────────────────────────────────────────


def emit_trend_seasonal(
    rng: np.random.Generator,
    n: int,
    L: int,
    *,
    hi_frac: float,
    exc_lo: float,
    exc_hi: float,
    clean_frac: float,
    clean_lo: float,
    clean_hi: float,
) -> np.ndarray:
    t = np.arange(L, dtype=np.float64)[None, :]
    level = rng.normal(0.0, 1.0, size=(n, 1))
    heavy = rng.random((n, 1)) < hi_frac
    slope = np.where(
        heavy,
        rng.normal(0.0, exc_hi, size=(n, 1)),
        rng.normal(0.0, exc_lo, size=(n, 1)),
    )
    tn = t / max(L - 1, 1)
    signal = level + slope * tn + _harmonics(rng, n, L, depth=3)
    # Clean mode: very low innov so periodic reconstruction is sharp.
    clean = rng.random(n) < clean_frac
    phi = rng.uniform(0.35, 0.92, size=n)
    sigma = np.where(
        clean[:, None],
        rng.uniform(clean_lo, clean_hi, size=(n, 1)),
        rng.uniform(0.06, 0.38, size=(n, 1)),
    )
    innov = rng.normal(0.0, 1.0, size=(n, L)) * sigma
    return signal + _ar1(innov, phi)


def emit_regime(rng: np.random.Generator, n: int, L: int) -> np.ndarray:
    level = np.cumsum(_jumps(rng, n, L, rate=2.0 / L, scale=2.0), axis=1)
    log_vol = np.cumsum(_jumps(rng, n, L, rate=2.0 / L, scale=0.4), axis=1)
    vol = np.exp(np.clip(log_vol, -3.0, 3.0)) * rng.uniform(0.08, 0.40, size=(n, 1))
    noise = rng.normal(0.0, 1.0, size=(n, L)) * vol
    seas = _harmonics(rng, n, L, depth=2) * rng.uniform(0.2, 1.0, size=(n, 1))
    return level + seas + noise


def emit_multiplicative(
    rng: np.random.Generator,
    n: int,
    L: int,
    *,
    hi_frac: float,
    exc_lo: float,
    exc_hi: float,
) -> np.ndarray:
    t = np.arange(L, dtype=np.float64)[None, :]
    heavy = rng.random((n, 1)) < hi_frac
    g = np.where(
        heavy,
        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 = np.exp(g * tn + rng.normal(0.0, 0.25, size=(n, 1)))
    amp = rng.uniform(0.15, 0.65, size=(n, 1))
    seas_shape = _harmonics(rng, n, L, depth=2)
    seas_sd = seas_shape.std(axis=1, keepdims=True)
    seas_shape = seas_shape / np.where(seas_sd < 1e-12, 1.0, seas_sd)
    seas = 1.0 + amp * seas_shape
    noise = 1.0 + rng.normal(0.0, 1.0, size=(n, L)) * rng.uniform(0.015, 0.10, size=(n, 1))
    scale = rng.uniform(1.0, 50.0, size=(n, 1))
    return scale * base * np.clip(seas, 0.05, None) * np.clip(noise, 0.05, None)


def emit_ar2(rng: np.random.Generator, n: int, L: int) -> np.ndarray:
    p1 = rng.uniform(0.45, 0.985, size=n)
    p2 = rng.uniform(-0.45, 0.45, size=n)
    a1 = p1 * (1.0 - p2)
    a2 = p2
    sigma = rng.uniform(0.15, 0.55, size=(n, 1))
    innov = rng.normal(0.0, 1.0, size=(n, L)) * sigma
    path = _ar2(innov, a1, a2)
    drift = rng.normal(0.0, 0.004, size=(n, 1)) * np.arange(L, dtype=np.float64)[None, :]
    return path + drift


def emit_integrated(rng: np.random.Generator, n: int, L: int) -> np.ndarray:
    twice = rng.random(n) < 0.30
    drift = rng.normal(0.0, 0.015, size=(n, 1))
    sigma = rng.uniform(0.15, 0.75, size=(n, 1))
    steps = rng.normal(0.0, 1.0, size=(n, L)) * sigma + drift
    walk = np.cumsum(steps, axis=1)
    walk2 = np.cumsum(walk, axis=1)
    return np.where(twice[:, None], walk2 / max(L, 1) ** 0.5, walk)


def emit_threshold(rng: np.random.Generator, n: int, L: int) -> np.ndarray:
    phi_pos = rng.uniform(0.4, 0.92, size=n)
    phi_neg = rng.uniform(-0.7, 0.35, size=n)
    c_pos = rng.normal(0.0, 0.25, size=n)
    c_neg = rng.normal(0.0, 0.25, size=n)
    sigma = rng.uniform(0.15, 0.55, size=(n, 1))
    innov = rng.normal(0.0, 1.0, size=(n, L)) * sigma
    return _jit_setar(
        np.ascontiguousarray(innov),
        np.ascontiguousarray(phi_pos),
        np.ascontiguousarray(phi_neg),
        np.ascontiguousarray(c_pos),
        np.ascontiguousarray(c_neg),
    )


def emit_chaotic(rng: np.random.Generator, n: int, L: int) -> np.ndarray:
    pick_sine = rng.random(n) < 0.55
    r_log = rng.uniform(3.4, 3.9, size=n)
    r_sin = rng.uniform(0.80, 0.97, size=n)
    x0 = rng.uniform(0.05, 0.95, size=n)
    return _jit_chaos(
        np.ascontiguousarray(pick_sine),
        np.ascontiguousarray(r_log),
        np.ascontiguousarray(r_sin),
        np.ascontiguousarray(x0),
        L,
    )


def emit_rff(rng: np.random.Generator, n: int, L: int, *, features: int = 24) -> np.ndarray:
    """Stationary GP via RFF; longer lengthscales → smoother, more forecastable."""
    t = np.arange(L, dtype=np.float64)[None, :]
    lengthscale = rng.uniform(40.0, 320.0, size=(n, 1))
    acc = np.zeros((n, L), dtype=np.float64)
    scale = np.sqrt(2.0 / features)
    for _ in range(features):
        omega = rng.normal(0.0, 1.0, size=(n, 1)) / lengthscale
        phase = rng.uniform(0.0, 2.0 * np.pi, size=(n, 1))
        np.add(acc, np.cos(omega * t + phase), out=acc)
    return scale * acc


def emit_intermittent(rng: np.random.Generator, n: int, L: int) -> np.ndarray:
    """Zero-inflated demand with mild weekly occurrence modulation."""
    t = np.arange(L, dtype=np.float64)[None, :]
    p0 = rng.uniform(0.06, 0.35, size=(n, 1))
    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)))
    )
    p = np.clip(p0 * week, 0.02, 0.55)
    occur = (rng.random((n, L)) < p).astype(np.float64)
    mag = rng.gamma(shape=2.0, scale=1.0, size=(n, L)) * rng.uniform(1.0, 8.0, size=(n, 1))
    floor = rng.uniform(0.0, 0.4, size=(n, 1))
    return floor + occur * mag


def emit_pulse(rng: np.random.Generator, n: int, L: int) -> np.ndarray:
    base = emit_rff(rng, n, L, features=24) * rng.uniform(0.5, 2.0, size=(n, 1))
    base += _harmonics(rng, n, L, depth=1) * rng.uniform(0.2, 1.0, size=(n, 1))
    spikes = _jumps(rng, n, L, rate=4.0 / L, scale=rng.uniform(2.5, 6.5, size=n))
    series = base + spikes
    hold = rng.random((n, L)) < (2.0 / L)
    hold[:, 0] = False
    _jit_holds(series, hold)
    return series


# ── weather: sliced archetype mixture (base / radiation / cloud) ────────────


def _cloud_cover(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, :]

    def _z_ar(phi_lo: float, phi_hi: float) -> np.ndarray:
        phi = rng.uniform(phi_lo, phi_hi, size=n)
        z = _ar1(rng.normal(0.0, 1.0, size=(n, L)), phi)
        return (z - z.mean(axis=1, keepdims=True)) / (z.std(axis=1, keepdims=True) + 1e-9)

    syn = _z_ar(0.990, 0.9990)
    mid = _z_ar(0.895, 0.95)
    fast = _z_ar(0.75, 0.87)

    g = rng.normal(0.0, 1.0, size=(n, L))
    spike = (rng.random((n, L)) < 0.05) * rng.uniform(5.0, 11.0, size=(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(axis=1, keepdims=True) + 1e-9)

    diur = np.sin(2.0 * np.pi * t / 24.0 + rng.uniform(0.0, 2.0 * np.pi, size=(n, 1)))
    u = (
        0.95 * rng.uniform(0.8, 1.2, size=(n, 1)) * syn
        + 0.82 * rng.uniform(0.7, 1.3, size=(n, 1)) * mid
        + 0.55 * rng.uniform(0.7, 1.3, size=(n, 1)) * fast
        + 0.30 * rng.uniform(0.7, 1.3, size=(n, 1)) * rough
        + 0.44 * rng.uniform(0.3, 1.4, size=(n, 1)) * diur
    )
    centre = np.clip(rng.normal(61.0, 29.0, size=(n, 1)), 2.0, 98.0)
    span = 57.0 * rng.uniform(0.82, 1.18, size=(n, 1))
    return np.round(np.clip(centre + span * u, 0.0, 100.0))


def emit_weather(rng: np.random.Generator, n: int, L: int) -> np.ndarray:
    """Mixture of diurnal base, night-floored radiation, and cloud cover."""
    t = np.arange(L, dtype=np.float64)[None, :]
    is_rad = rng.random(n) < 0.25
    is_cloud = (rng.random(n) < 0.24) & ~is_rad
    is_wbp = ~is_cloud
    out = np.empty((n, L), dtype=np.float64)

    wbp = np.nonzero(is_wbp)[0]
    m = wbp.size
    if m:
        rad_rows = is_rad[wbp][:, 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)
        weekly_on = (rng.random((m, 1)) < 0.45).astype(np.float64)
        seas += weekly_on * 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_on = (rng.random((m, 1)) < 0.35).astype(np.float64)
        seas += month_on * 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))
        )
        y_amp = rng.uniform(0.25, 1.6, size=(m, 1))
        y_per = rng.uniform(2000.0, 9000.0, size=(m, 1))
        yearly = y_amp * np.sin(2.0 * np.pi * t / y_per + 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(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.85, 1.65, size=(m, 1))
        diurnal = ramp * np.maximum(a1 * np.sin(2.0 * np.pi * t / 24.0 + p1) - thr, 0.0)
        night = diurnal <= 0.0
        rad = level + diurnal + yearly + np.where(night, noise * 0.12, noise)
        out[wbp] = np.where(rad_rows, rad, base)

    cloud = np.nonzero(is_cloud)[0]
    if cloud.size:
        out[cloud] = _cloud_cover(rng, int(cloud.size), L)
    return out


# ── Grammar-of-Time productions ─────────────────────────────────────────────
# Lightweight stems only — compositions draw several stems per row, so keep
# them FFT/AR/seasonal (no per-t Python loops beyond the shared AR1 scan).


def _stem_ar_seasonal(rng: np.random.Generator, n: int, L: int) -> np.ndarray:
    seas = _harmonics(rng, n, L, depth=2)
    phi = rng.uniform(0.15, 0.90, size=n)
    sigma = rng.uniform(0.12, 0.55, 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(innov, phi))


def _stem_rff(rng: np.random.Generator, n: int, L: int) -> np.ndarray:
    return _row_standardize(emit_rff(rng, n, L, features=20))


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 _stem_ar2(rng: np.random.Generator, n: int, L: int) -> np.ndarray:
    return _row_standardize(emit_ar2(rng, n, L))


def _stem_weatherish(rng: np.random.Generator, n: int, L: int) -> np.ndarray:
    """Cheap diurnal+weekly stem (no cloud path) for GoT weather phrases."""
    t = np.arange(L, dtype=np.float64)[None, :]
    a1 = rng.uniform(0.5, 2.0, size=(n, 1))
    p1 = rng.uniform(0.0, 2.0 * np.pi, size=(n, 1))
    a2 = rng.uniform(0.1, 0.6, size=(n, 1))
    p2 = rng.uniform(0.0, 2.0 * np.pi, size=(n, 1))
    seas = a1 * np.sin(2.0 * np.pi * t / 24.0 + p1) + a2 * np.sin(2.0 * np.pi * t / 168.0 + p2)
    phi = rng.uniform(0.5, 0.92, size=n)
    sigma = rng.uniform(0.05, 0.25, size=(n, 1))
    return _row_standardize(seas + _ar1(rng.normal(0.0, 1.0, size=(n, L)) * sigma, phi))


def _sample_stem(rng: np.random.Generator, n: int, L: int) -> np.ndarray:
    builders = (
        _stem_ar_seasonal,
        _stem_rff,
        _stem_integrated,
        _stem_ar2,
        _stem_weatherish,
    )
    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] = builder(rng, int(idx.size), L)
    return out


def emit_got_compose(
    rng: np.random.Generator,
    n: int,
    L: int,
    *,
    depth: int,
    mul_frac: float,
    hi_frac: float,
    exc_lo: float,
    exc_hi: float,
) -> np.ndarray:
    """Production: Stem ⊕ Stem [⊕ Stem] — additive or multiplicative phrase."""
    depth = int(np.clip(depth, 2, 4))
    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))
        add_mask = active & (~use_mul[:, None])
        out = np.where(add_mask, out + w * stem, out)
        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)

    t = np.arange(L, dtype=np.float64)[None, :] / max(L - 1, 1)
    heavy = rng.random((n, 1)) < hi_frac
    exc = np.where(
        heavy,
        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

    punct = rng.random(n) < 0.35
    if punct.any():
        jumps = np.cumsum(_jumps(rng, n, L, rate=2.5 / L, scale=1.5), axis=1)
        out[punct] = out[punct] + jumps[punct]
    return _rescale(out, rng)


def emit_got_splice(
    rng: np.random.Generator,
    n: int,
    L: int,
    *,
    n_cuts: int,
    hi_frac: float,
    exc_lo: float,
    exc_hi: float,
) -> np.ndarray:
    """Production: Stem ‖ Stem — clause boundaries splice different dynamics."""
    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)]
    cuts = np.sort(
        rng.integers(max(1, L // 8), max(2, (7 * L) // 8), size=(n, n_cuts)),
        axis=1,
    )
    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)

    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)
        near = np.abs(dist) <= blend_w
        blended = (1.0 - gate) * clauses[c] + gate * clauses[c + 1]
        out = np.where(near, blended, out)

    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)
    heavy = rng.random((n, 1)) < hi_frac
    exc = np.where(
        heavy,
        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
    return _rescale(out, rng)


def emit_got_nested(
    rng: np.random.Generator,
    n: int,
    L: int,
    *,
    nest_ratio: float,
) -> np.ndarray:
    """Production: Envelope ⋉ Carrier — slow scale nests a fast carrier."""
    nest_ratio = float(np.clip(nest_ratio, 2.0, 24.0))
    t = np.arange(L, dtype=np.float64)[None, :]

    use_rff_env = rng.random(n) < 0.55
    env = np.empty((n, L), dtype=np.float64)
    rff_rows = np.nonzero(use_rff_env)[0]
    sin_rows = np.nonzero(~use_rff_env)[0]
    if rff_rows.size:
        # Long-lengthscale RFF as envelope (force smooth).
        env[rff_rows] = _row_standardize(
            emit_rff(rng, int(rff_rows.size), L, features=16)
        )
    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)

    carrier = _row_standardize(_harmonics(rng, n, L, depth=3))
    mix_ar = rng.random(n) < 0.45
    if mix_ar.any():
        ar = _row_standardize(emit_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
    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)

    sigma = rng.uniform(0.05, 0.30, 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(innov, phi)
    return _rescale(out, rng)


def _delay_batch(x: np.ndarray, lags: np.ndarray) -> np.ndarray:
    """Causal delay with edge hold, vectorised over unique lag values."""
    n, L = x.shape
    out = np.empty_like(x)
    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 emit_got_causal(
    rng: np.random.Generator,
    n: int,
    L: int,
    *,
    lag_frac: float,
) -> np.ndarray:
    """Production: Driver ▷ Response — lagged temporal causal chain."""
    lag_frac = float(np.clip(lag_frac, 0.01, 0.25))
    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.45, size=(n, 1))
    innov = rng.normal(0.0, 1.0, size=(n, L)) * sigma
    resp = resp + _ar1(innov, phi)
    seas_on = rng.random((n, 1)) < 0.5
    resp = resp + seas_on * _harmonics(rng, n, L, depth=2) * rng.uniform(0.1, 0.8, size=(n, 1))
    return _rescale(resp, rng)


# ── generator entry point ───────────────────────────────────────────────────


class Generator(DataGenerator):
    """Weather + Grammar-of-Time mixture. Submit as ``generator.Generator``."""

    def __init__(self, config_dir: str, *, seed: int) -> None:
        cfg_path = Path(config_dir) / "config.json"
        cfg: dict[str, Any] = (
            json.loads(cfg_path.read_text(encoding="utf-8")) if cfg_path.is_file() else {}
        )
        self._cfg = cfg
        self._seed = int(seed)
        self._min_len = int(cfg.get("min_length", 64))
        self._max_len = int(cfg.get("max_length", 4096))
        if self._min_len < 1 or self._max_len < self._min_len:
            raise ValueError(f"invalid length band [{self._min_len}, {self._max_len}]")

        mass = dict(_PRIOR_MASS)
        for key, val in dict(cfg.get("family_weights", {})).items():
            if key in mass:
                mass[key] = float(val)
        weights = np.asarray([mass[name] for name in _PRIOR_NAMES], dtype=np.float64)
        if not np.all(np.isfinite(weights)) or weights.min() < 0 or weights.sum() <= 0:
            raise ValueError("family_weights must be finite, non-negative, and not all zero")
        self._weights = weights / weights.sum()

        self._tr_hi = float(cfg.get("tr_hi_frac", 0.20))
        self._tr_lo = float(cfg.get("tr_exc_lo", 0.35))
        self._tr_hi_s = float(cfg.get("tr_exc_hi", 2.0))
        self._gr_lo = float(cfg.get("gr_exc_lo", 0.25))
        self._gr_hi = float(cfg.get("gr_exc_hi", 1.4))
        self._clean_frac = float(cfg.get("sa_clean_frac", 0.35))
        self._clean_lo = float(cfg.get("sa_clean_lo", 0.02))
        self._clean_hi = float(cfg.get("sa_clean_hi", 0.12))
        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._fixed = self._min_len == self._max_len

    @property
    def name(self) -> str:
        return str(self._cfg.get("name", "zenfro-v2-grammar-of-time"))

    def _builders(self) -> tuple[Callable[..., np.ndarray], ...]:
        return (
            lambda rng, n, L: emit_trend_seasonal(
                rng,
                n,
                L,
                hi_frac=self._tr_hi,
                exc_lo=self._tr_lo,
                exc_hi=self._tr_hi_s,
                clean_frac=self._clean_frac,
                clean_lo=self._clean_lo,
                clean_hi=self._clean_hi,
            ),
            emit_regime,
            lambda rng, n, L: emit_multiplicative(
                rng, n, L, hi_frac=self._tr_hi, exc_lo=self._gr_lo, exc_hi=self._gr_hi
            ),
            emit_ar2,
            emit_integrated,
            emit_threshold,
            emit_chaotic,
            emit_rff,
            emit_intermittent,
            emit_pulse,
            emit_weather,
            lambda rng, n, L: emit_got_compose(
                rng,
                n,
                L,
                depth=self._got_depth,
                mul_frac=self._got_mul_frac,
                hi_frac=self._tr_hi,
                exc_lo=self._tr_lo,
                exc_hi=self._tr_hi_s,
            ),
            lambda rng, n, L: emit_got_splice(
                rng,
                n,
                L,
                n_cuts=self._got_splice_cuts,
                hi_frac=self._tr_hi,
                exc_lo=self._tr_lo,
                exc_hi=self._tr_hi_s,
            ),
            lambda rng, n, L: emit_got_nested(rng, n, L, nest_ratio=self._got_nest_ratio),
            lambda rng, n, L: emit_got_causal(
                rng, n, L, lag_frac=self._got_causal_lag_frac
            ),
        )

    def generate(self, n_series: int) -> Iterator[np.ndarray]:
        if n_series <= 0:
            return
        rng = np.random.default_rng(self._seed)
        builders = self._builders()
        L_max = self._max_len
        done = 0
        while done < n_series:
            # Always draw a full batch so series i is a pure function of (seed, i).
            lengths = rng.integers(self._min_len, L_max + 1, size=_BATCH)
            picks = rng.choice(len(_PRIOR_NAMES), size=_BATCH, p=self._weights)
            slots: list[np.ndarray | None] = [None] * _BATCH
            for fam_id, build in enumerate(builders):
                idx = np.nonzero(picks == fam_id)[0]
                if idx.size == 0:
                    continue
                block = _finite(build(rng, int(idx.size), L_max))
                if self._fixed:
                    for row, slot in enumerate(idx):
                        slots[int(slot)] = np.ascontiguousarray(block[row], dtype=np.float64)
                else:
                    for row, slot in enumerate(idx):
                        L = int(lengths[slot])
                        slots[int(slot)] = np.ascontiguousarray(block[row, :L], dtype=np.float64)
            take = min(_BATCH, n_series - done)
            for arr in slots[:take]:
                if arr is None:  # pragma: no cover
                    raise RuntimeError("unfilled series slot")
                yield arr
            done += take