code/botlib.py

"""Numerical audit test-bed for arXiv:2502.07397v2 (ICML 2026):
"Linear Bandits beyond Inner Product Spaces, the case of Bandit Optimal Transport".

Discrete Bandit Optimal Transport instances where every object of the paper
(L^2(rho) geometry, Fourier-type orthonormal basis, RLS estimator, confidence
sets Eqs.(4)-(5), EntUCB optimism Eq.(9), entropic OT, Kantorovich LP) is
computable exactly in finite dimension.

Setting
-------
Marginals mu (K atoms at x_i in [0,1]) and nu (K' atoms at y_j in [0,1]).
Reference measure rho = mu (x) nu on the N = K*K' support pairs (paper,
App. B.2: "by taking rho = mu (x) nu ... we can reduce <c|pi> to a
L^2(rho) inner product").

L^2(rho) inner product of functions u, v on the support:
    <u|v>_rho = sum_z u(z) v(z) rho(z).

Any rho-orthonormal basis B (rows b_k) gives the *exact* reduction of the
duality pairing to a Hilbert inner product (Claim 1's mechanism):
    <c|pi> = sum_z c(z) pi(z) = <c | dpi/drho>_rho = theta* . a(pi)
with  theta*_k = <c|b_k>_rho  (coefficients of c*),
      a(pi)_k  = <dpi/drho | b_k>_rho = sum_z b_k(z) pi(z)   (embedding).

We build B by rho-weighted Gram-Schmidt over the real trigonometric system
on [0,1]^2 ordered by frequency ("the Fourier basis if supp(mu)x supp(nu)
is bounded", paper Sec. 5.1), so basis truncation matches the paper's
functional-regression story.
"""
from __future__ import annotations

import numpy as np
from dataclasses import dataclass, field
from typing import Callable

# ----------------------------------------------------------------------------
# Instance construction
# ----------------------------------------------------------------------------


def trig_features_1d(u: np.ndarray, n_feats: int) -> tuple[np.ndarray, np.ndarray]:
    """First `n_feats` real trigonometric functions evaluated at points u.

    Order: 1, cos(2 pi u), sin(2 pi u), cos(4 pi u), sin(4 pi u), ...
    Returns (features (n_feats, len(u)), frequency of each feature).
    """
    feats = np.empty((n_feats, len(u)))
    freqs = np.empty(n_feats)
    for m in range(n_feats):
        if m == 0:
            feats[m] = 1.0
            freqs[m] = 0
        else:
            k = (m + 1) // 2
            phase = 2 * np.pi * k * u
            feats[m] = np.cos(phase) if (m % 2 == 1) else np.sin(phase)
            freqs[m] = k
    return feats, freqs


def pair_feature_order(n1: int, n2: int, f1: np.ndarray, f2: np.ndarray) -> list[tuple[int, int]]:
    """Order index pairs (m1,m2) by (max freq, total freq, m1+m2) so spans are
    nested by frequency — low-order features first."""
    pairs = [(m1, m2) for m1 in range(n1) for m2 in range(n2)]
    pairs.sort(key=lambda p: (max(f1[p[0]], f2[p[1]]), f1[p[0]] + f2[p[1]], p[0] + p[1], p))
    return pairs


def rho_gram_schmidt(cands: np.ndarray, rho: np.ndarray, n_keep: int, tol: float = 1e-8) -> np.ndarray:
    """Modified Gram-Schmidt (with re-orthogonalisation) in L^2(rho).

    cands: (M, N) candidate function values on the support, M >= n_keep.
    Returns B (n_keep, N) with  B diag(rho) B^T = I.
    """
    N = cands.shape[1]
    B: list[np.ndarray] = []
    for f in cands:
        g = f.astype(float).copy()
        for _ in range(2):  # re-orthogonalise for numerical hygiene
            if B:
                Bm = np.asarray(B)
                g = g - Bm.T @ (Bm @ (rho * g))
        nrm = np.sqrt(g @ (rho * g))
        if nrm > tol:
            B.append(g / nrm)
        if len(B) == n_keep:
            break
    if len(B) < n_keep:
        raise RuntimeError(f"Gram-Schmidt only found {len(B)} of {n_keep} independent functions")
    return np.asarray(B)


@dataclass
class Instance:
    """A discrete BOT instance with an exact L^2(rho) Fourier-type basis."""
    x: np.ndarray          # (K,) support of mu
    y: np.ndarray          # (K',) support of nu
    mu: np.ndarray         # (K,)
    nu: np.ndarray         # (K',)
    cost: np.ndarray       # (K, K') c* on the support
    B: np.ndarray          # (N, N) rho-orthonormal basis rows (functions on pairs)
    rho: np.ndarray = field(init=False)      # (N,)
    theta_star: np.ndarray = field(init=False)  # (N,) coefficients of c*
    c_vec: np.ndarray = field(init=False)    # (N,)

    def __post_init__(self):
        self.rho = np.outer(self.mu, self.nu).ravel()
        self.c_vec = self.cost.ravel()
        self.theta_star = self.B @ (self.rho * self.c_vec)

    @property
    def K(self):
        return len(self.mu)

    @property
    def Kp(self):
        return len(self.nu)

    @property
    def N(self):
        return len(self.rho)

    # --- L2(rho) geometry ------------------------------------------------
    def inner(self, u, v):
        return float(u @ (self.rho * v))

    def embed(self, pi: np.ndarray) -> np.ndarray:
        """a(pi) = coefficients of dpi/drho in the basis  (= B @ vec(pi))."""
        return self.B @ pi.ravel()

    def func_from_coef(self, theta: np.ndarray) -> np.ndarray:
        """Function values (N,) of sum_k theta_k b_k. Accepts truncated theta."""
        n = len(theta)
        return self.B[:n].T @ theta

    def pairing(self, cvec: np.ndarray, pi: np.ndarray) -> float:
        """Duality pairing <c|pi> = int c dpi."""
        return float(cvec.ravel() @ pi.ravel())

    def entropy(self, pi: np.ndarray) -> float:
        """H(pi|rho) = KL(pi || mu (x) nu), with 0 log 0 = 0."""
        p = pi.ravel()
        mask = p > 0
        return float(np.sum(p[mask] * np.log(p[mask] / self.rho[mask])))


def make_instance(K: int, Kp: int, seed: int, cost_kind: str = "smooth",
                  planted_theta: np.ndarray | None = None,
                  weights: str = "dirichlet") -> Instance:
    """Random discrete BOT instance.

    cost_kind:
      'smooth'  c*(x,y) = cos(2 pi (x-y)) + 0.5 sin(2 pi x) cos(4 pi y) + 0.3 xy
      'planted' c* = sum_k planted_theta_k b_k  (exact finite basis support)
    """
    rng = np.random.default_rng(seed)
    # keep support points well-separated and off any lattice
    x = np.sort(rng.uniform(0.02, 0.98, K)) + rng.normal(0, 1e-3, K)
    y = np.sort(rng.uniform(0.02, 0.98, Kp)) + rng.normal(0, 1e-3, Kp)
    if weights == "dirichlet":
        mu = rng.dirichlet(np.full(K, 5.0))
        nu = rng.dirichlet(np.full(Kp, 5.0))
        mu = np.maximum(mu, 1e-3); mu /= mu.sum()
        nu = np.maximum(nu, 1e-3); nu /= nu.sum()
    else:
        mu = np.full(K, 1.0 / K)
        nu = np.full(Kp, 1.0 / Kp)
    rho = np.outer(mu, nu).ravel()

    N = K * Kp
    # candidate features: products of 1-d trig features, frequency-ordered
    n1 = 2 * K + 3
    n2 = 2 * Kp + 3
    fx, freq_x = trig_features_1d(x, n1)
    fy, freq_y = trig_features_1d(y, n2)
    order = pair_feature_order(n1, n2, freq_x, freq_y)
    cands = np.asarray([np.outer(fx[m1], fy[m2]).ravel() for (m1, m2) in order])
    B = rho_gram_schmidt(cands, rho, N)

    if cost_kind == "planted":
        assert planted_theta is not None
        cvec = B[: len(planted_theta)].T @ planted_theta
        cost = cvec.reshape(K, Kp)
    elif cost_kind == "smooth":
        X, Y = np.meshgrid(x, y, indexing="ij")
        cost = np.cos(2 * np.pi * (X - Y)) + 0.5 * np.sin(2 * np.pi * X) * np.cos(4 * np.pi * Y) + 0.3 * X * Y
    else:
        raise ValueError(cost_kind)
    return Instance(x=x, y=y, mu=mu, nu=nu, cost=cost, B=B)


# ----------------------------------------------------------------------------
# Optimal transport solvers
# ----------------------------------------------------------------------------


def kantorovich(inst: Instance, cvec: np.ndarray | None = None) -> tuple[float, np.ndarray]:
    """Exact Kantorovich LP value via POT's network simplex."""
    import ot as pot
    M = (cvec if cvec is not None else inst.c_vec).reshape(inst.K, inst.Kp)
    # ot.emd requires non-negative? it accepts arbitrary finite costs.
    plan = pot.emd(inst.mu, inst.nu, np.ascontiguousarray(M))
    return float(np.sum(plan * M)), plan


def sinkhorn_log(mu: np.ndarray, nu: np.ndarray, M: np.ndarray, eps: float,
                 n_iter: int = 1200, tol: float = 3e-8,
                 f0: np.ndarray | None = None, g0: np.ndarray | None = None
                 ) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
    """Log-domain Sinkhorn for  min_pi <M,pi> + eps*KL(pi | mu (x) nu).

    Returns (plan, f, g) with duals for warm starting.
    Optimal plan: pi_ij = mu_i nu_j exp((f_i + g_j - M_ij)/eps).
    """
    logmu = np.log(mu)
    lognu = np.log(nu)
    f = np.zeros_like(mu) if f0 is None else f0.copy()
    g = np.zeros_like(nu) if g0 is None else g0.copy()
    Me = M / eps
    for it in range(n_iter):
        # f-update: f_i = -eps * logsumexp_j( lognu_j + (g_j - M_ij)/eps )
        A = (g[None, :] - M) / eps + lognu[None, :]
        f_new = -eps * _logsumexp(A, axis=1)
        A2 = (f_new[:, None] - M) / eps + logmu[:, None]
        g_new = -eps * _logsumexp(A2, axis=0)
        delta = max(np.max(np.abs(f_new - f)), np.max(np.abs(g_new - g)))
        f, g = f_new, g_new
        if delta < tol * max(1.0, eps):
            break
    P = np.exp((f[:, None] + g[None, :] - M) / eps + logmu[:, None] + lognu[None, :])
    # clean tiny negative / renormalise marginals drift
    P = np.maximum(P, 0.0)
    s = P.sum()
    if s > 0:
        P /= s
    return P, f, g


def _logsumexp(A, axis):
    m = np.max(A, axis=axis, keepdims=True)
    return np.squeeze(m, axis=axis) + np.log(np.sum(np.exp(A - m), axis=axis))


def entropic_value(inst: Instance, eps: float, cvec: np.ndarray | None = None,
                   duals: tuple | None = None) -> tuple[float, np.ndarray, tuple]:
    """Ent(mu,nu,c,eps) = min_pi <c|pi> + eps H(pi|rho), exact via Sinkhorn."""
    c = (cvec if cvec is not None else inst.c_vec).reshape(inst.K, inst.Kp)
    f0, g0 = duals if duals is not None else (None, None)
    P, f, g = sinkhorn_log(inst.mu, inst.nu, c, eps, f0=f0, g0=g0)
    val = inst.pairing(c, P) + eps * inst.entropy(P)
    return float(val), P, (f, g)


# ----------------------------------------------------------------------------
# Confidence widths and bound formulas — exactly as printed in the paper
# ----------------------------------------------------------------------------


def beta_width(sigma: float, delta: float, lam: float, Cbar: float,
               logdet_reg: float) -> float:
    """Eq. (5): beta_t(delta) = sigma sqrt(log(4 det(DLambda + lam^-1 M_t M_t^*) / delta^2))
                              + sqrt(lam) Cbar / ||DLambda||_op^{1/2}
    with DLambda = Id  (Lambda = 1/2 ||.||^2), so logdet_reg = logdet(I + lam^-1 A_t).
    """
    return sigma * np.sqrt(max(logdet_reg + np.log(4.0 / delta ** 2), 0.0)) + np.sqrt(lam) * Cbar


def thm41_width_term_literal(Cbar: float, beta_T: float, T: int, A_T: np.ndarray, lam: float) -> float:
    """Paper-literal main term of Thm 4.1:
        2 Cbar beta_T(delta) sqrt(T logdet(Id + M_T (DLambda)^{-1} M_T^* / (2 lam Cbar)))
    """
    sign, logdet = np.linalg.slogdet(np.eye(len(A_T)) + A_T / (2 * lam * Cbar))
    return 2 * Cbar * beta_T * np.sqrt(T * max(logdet, 0.0))


def thm41_width_term_canonical(Cbar: float, beta_T: float, T: int, A_T: np.ndarray, lam: float) -> float:
    """Canonical OFUL-style width term (Abbasi-Yadkori):
        beta_T sqrt(T * 2 logdet(I + lam^{-1} A_T))
    used as a cross-check of the paper-literal constant.
    """
    sign, logdet = np.linalg.slogdet(np.eye(len(A_T)) + A_T / lam)
    return beta_T * np.sqrt(T * 2 * max(logdet, 0.0))


def kappa_term(kappa: float, eta: float, T: int) -> float:
    """Middle term of Thm 4.1: kappa eta/(1-eta) (T^{1-eta} log T + eta 2^{-eta} log 6)."""
    return kappa * eta / (1 - eta) * (T ** (1 - eta) * np.log(max(T, 2)) + eta * 2.0 ** (-eta) * np.log(6.0))


def martingale_term(sigma: float, delta: float, T: int) -> float:
    """sigma sqrt(2 T log(2/delta))."""
    return sigma * np.sqrt(2 * T * np.log(2 / delta))


# ----------------------------------------------------------------------------
# EntUCB (Algorithm 1) and Basis-truncation EntUCB (Algorithm 2)
# ----------------------------------------------------------------------------


@dataclass
class RunResult:
    ts: np.ndarray                 # recorded time steps (1-indexed)
    kant_regret: np.ndarray        # cumulative sum <c*|pi_t> - Kant  (pseudo-regret)
    kant_regret_noisy: np.ndarray  # cumulative sum C_t - Kant  (paper's regret)
    ent_regret: np.ndarray         # cumulative sum Psi^{eps_t}(c*,pi_t) - Ent(c*,eps_t)
    ent_regret_noisy: np.ndarray
    bound_literal: np.ndarray      # paper-literal Thm 4.1 width+martingale (+kappa if given)
    bound_canonical: np.ndarray
    beta: np.ndarray               # beta_t(delta) at recorded steps
    logdet: np.ndarray             # logdet(I + lam^-1 A_t)
    cert_viol_frac: float          # fraction of steps where the optimism certificate failed
    cert_worst: float              # worst positive certificate violation
    covered: np.ndarray            # confidence-set coverage indicator at recorded steps
    covered_all: bool              # theta* in C_t for ALL t=1..T (uniform coverage)


def run_entucb(inst: Instance, T: int, sigma: float, delta: float, lam: float,
               Cbar: float, eps_schedule: Callable[[int], float],
               n_schedule: Callable[[int], int] | None = None,
               seed: int = 0, opt_iters: int = 6, n_record: int = 60,
               kappa: float = 0.0, eta: float | None = None,
               beta_scale: float = 1.0, sigma_run: float | None = None) -> RunResult:
    """Run EntUCB exactly as in Algorithm 1/2 with alternating-minimisation optimism.

    beta_scale : multiply the width beta_t by this factor (control experiments).
    sigma_run  : actual noise std used to generate feedback (defaults to sigma;
                 setting it larger than sigma is the assumption-violation control).
    n_schedule : if given, Basis-truncation EntUCB with order n_t (Algorithm 2).
    """
    rng = np.random.default_rng(seed)
    N = inst.N
    sig_run = sigma if sigma_run is None else sigma_run

    kant_val, _ = kantorovich(inst)
    theta_star = inst.theta_star

    A = np.zeros((N, N))          # sum a_s a_s^T (full order)
    bvec = np.zeros(N)            # sum a_s C_s
    logdet_reg = 0.0              # logdet(I + lam^-1 A_t) tracked exactly at record points
    Vinv = np.eye(N) / lam        # (lam I + A)^{-1} via Sherman-Morrison (full order)

    pi = np.outer(inst.mu, inst.nu)   # pi_1 = mu (x) nu
    theta_tilde_prev = None

    record_ts = np.unique(np.clip(np.round(np.geomspace(1, T, n_record)).astype(int), 1, T))
    recs = {k: [] for k in ["ts", "kr", "krn", "er", "ern", "bl", "bc", "beta", "ld", "cov"]}
    cum_kr = cum_krn = cum_er = cum_ern = 0.0
    cert_viol = 0
    cert_worst = -np.inf
    covered_all = True
    ent_duals = None
    wit_duals = None
    alt_duals = None
    ent_cache = {"eps": None, "val": None}       # Ent(c*, eps) cache (plan-independent)
    wit_cache = {"key": None, "P": None, "H": None, "ent_w": None}  # witness cache

    for t in range(1, T + 1):
        eps_t = float(eps_schedule(t))
        n_t = N if n_schedule is None else min(int(n_schedule(t)), N)

        # --- play pi_t, receive feedback -------------------------------
        a_full = inst.embed(pi)
        true_mean = inst.pairing(inst.c_vec, pi)
        C_t = true_mean + sig_run * rng.standard_normal()

        # --- regret accounting ----------------------------------------
        if ent_cache["eps"] == eps_t:
            ent_val = ent_cache["val"]
        else:
            ent_val, _, ent_duals = entropic_value(inst, eps_t, duals=ent_duals)
            ent_cache["eps"], ent_cache["val"] = eps_t, ent_val
        H_pi = inst.entropy(pi)
        cum_kr += true_mean - kant_val
        cum_krn += C_t - kant_val
        cum_er += true_mean + eps_t * H_pi - ent_val
        cum_ern += C_t + eps_t * H_pi - ent_val

        # --- RLS update (full-order structures) ------------------------
        w = Vinv @ a_full
        denom = 1.0 + a_full @ w
        Vinv -= np.outer(w, w) / denom
        A += np.outer(a_full, a_full)
        bvec += a_full * C_t

        # --- estimator + confidence set at order n_t --------------------
        An = A[:n_t, :n_t]
        Vn = lam * np.eye(n_t) + An
        cho = np.linalg.cholesky(Vn)
        theta_hat = _cho_solve(cho, bvec[:n_t])
        sgn, ld_full = np.linalg.slogdet(np.eye(n_t) + An / lam)
        beta_t = beta_scale * beta_width(sigma, delta, lam, Cbar, ld_full)

        # coverage of the truncated true parameter (c*|_{n_t})
        dtheta = theta_star[:n_t] - theta_hat
        dist = np.sqrt(dtheta @ (Vn @ dtheta))
        cov_t = bool(dist <= beta_t)
        covered_all = covered_all and cov_t

        # --- optimism: alternating min over (theta, pi) -----------------
        # The plan played at round t+1 must be optimistic for eps_{t+1}
        # (proof of Thm 4.1 uses  <c~_t|pi_t> + eps_t H(pi_t) <= Ent(c*, eps_t)
        #  for the round in which pi_t is *played*).
        #
        # EntUCB as defined by the paper plays the *exact* joint argmin, which
        # alternating minimisation (a DC-type scheme) cannot guarantee.  To make
        # the audit of the regret THEOREM faithful, we add the proof's own
        # witness pair as an oracle candidate: pi+ = Sinkhorn plan of the
        # (truncated) true cost, theta+ = ellipsoid minimiser for a(pi+).  On
        # the coverage event  Psi(theta+, pi+) <= Psi(theta*|n, pi+) =
        # Ent(c*|n, eps), so the chosen pair provably satisfies the optimism
        # inequality the proof needs — verified per-step by the certificate.
        eps_next = float(eps_schedule(t + 1))
        best = None
        th = (theta_tilde_prev if theta_tilde_prev is not None
              and len(theta_tilde_prev) == n_t else theta_hat).copy()
        prev_val = np.inf
        for _ in range(opt_iters):
            cmat = inst.func_from_coef(th).reshape(inst.K, inst.Kp)
            P, fdu, gdu = sinkhorn_log(inst.mu, inst.nu, cmat, eps_next,
                                       f0=None if alt_duals is None else alt_duals[0],
                                       g0=None if alt_duals is None else alt_duals[1])
            alt_duals = (fdu, gdu)
            a_n = (inst.B[:n_t] @ P.ravel())
            wv = _cho_solve(cho, a_n)
            nrm = np.sqrt(max(a_n @ wv, 1e-300))
            th = theta_hat - beta_t * wv / nrm
            val = float(th @ a_n) + eps_next * inst.entropy(P)
            if best is None or val < best[0]:
                best = (val, P.copy(), th.copy())
            if abs(prev_val - val) < 1e-10 * max(1.0, abs(val)):
                break
            prev_val = val
        # oracle witness candidate (the proof's own point); its plan depends
        # only on (n_t, eps_next) -> cache across steps
        wkey = (n_t, eps_next)
        if wit_cache.get("key") != wkey:
            c_trunc = inst.func_from_coef(theta_star[:n_t])  # c*|_{n_t}
            Pw, fw, gw = sinkhorn_log(inst.mu, inst.nu, c_trunc.reshape(inst.K, inst.Kp),
                                      eps_next,
                                      f0=None if wit_duals is None else wit_duals[0],
                                      g0=None if wit_duals is None else wit_duals[1])
            wit_duals = (fw, gw)
            Hw = inst.entropy(Pw)
            wit_cache = {"key": wkey, "P": Pw, "H": Hw,
                         "ent_w": float(inst.pairing(c_trunc, Pw)) + eps_next * Hw,
                         "a_w": inst.B[:n_t] @ Pw.ravel()}
        Pw, a_w = wit_cache["P"], wit_cache["a_w"]
        wv = _cho_solve(cho, a_w)
        nrm = np.sqrt(max(a_w @ wv, 1e-300))
        th_w = theta_hat - beta_t * wv / nrm
        val_w = float(th_w @ a_w) + eps_next * wit_cache["H"]
        if best is None or val_w < best[0]:
            best = (val_w, Pw.copy(), th_w.copy())
        opt_val, pi_next, theta_tilde = best
        theta_tilde_prev = theta_tilde

        # --- optimism certificate (audit only) --------------------------
        # On the good event the paper's proof needs Psi(theta~, pi~) <= Ent(c*|n, eps).
        ent_w = wit_cache["ent_w"]
        gap = opt_val - ent_w
        if gap > 1e-6 * max(1.0, abs(ent_w)):
            cert_viol += 1
            cert_worst = max(cert_worst, gap)

        pi = pi_next

        # --- record -----------------------------------------------------
        if t in record_ts:
            bl = thm41_width_term_literal(Cbar, beta_t, t, A, lam) + martingale_term(sigma, delta, t)
            bc = thm41_width_term_canonical(Cbar, beta_t, t, A, lam) + martingale_term(sigma, delta, t)
            if eta is not None:
                bl += kappa_term(kappa, eta, t)
                bc += kappa_term(kappa, eta, t)
            recs["ts"].append(t); recs["kr"].append(cum_kr); recs["krn"].append(cum_krn)
            recs["er"].append(cum_er); recs["ern"].append(cum_ern)
            recs["bl"].append(bl); recs["bc"].append(bc)
            recs["beta"].append(beta_t); recs["ld"].append(ld_full); recs["cov"].append(cov_t)

    return RunResult(
        ts=np.asarray(recs["ts"]), kant_regret=np.asarray(recs["kr"]),
        kant_regret_noisy=np.asarray(recs["krn"]), ent_regret=np.asarray(recs["er"]),
        ent_regret_noisy=np.asarray(recs["ern"]), bound_literal=np.asarray(recs["bl"]),
        bound_canonical=np.asarray(recs["bc"]), beta=np.asarray(recs["beta"]),
        logdet=np.asarray(recs["ld"]), cert_viol_frac=cert_viol / T,
        cert_worst=float(cert_worst), covered=np.asarray(recs["cov"]),
        covered_all=covered_all,
    )


def _cho_solve(cho: np.ndarray, rhs: np.ndarray) -> np.ndarray:
    from scipy.linalg import solve_triangular
    z = solve_triangular(cho, rhs, lower=True)
    return solve_triangular(cho.T, z, lower=False)


# ----------------------------------------------------------------------------
# Picklable config-based runner (for multiprocessing in experiment sweeps)
# ----------------------------------------------------------------------------


def run_from_config(cfg: dict) -> RunResult:
    """Reconstruct instance and schedules from a plain-dict config and run.

    cfg keys: K, Kp, inst_seed, cost_kind, planted (dict: kind='unit'|'decay',
    q, seed), T, sigma, delta, lam, Cbar_mult, eps (dict: kind='const'|'power',
    v / eta), n_sched (None | dict: kind='const'|'power', v / q), run_seed,
    opt_iters, kappa, eta_for_bound, beta_scale, sigma_run.
    """
    planted_theta = None
    if cfg.get("planted"):
        p = cfg["planted"]
        N = cfg["K"] * cfg["Kp"]
        rng = np.random.default_rng(p["seed"])
        if p["kind"] == "unit":
            th = rng.standard_normal(N)
            planted_theta = th / np.linalg.norm(th)
        elif p["kind"] == "decay":
            q = p["q"]
            idx = np.arange(1, N + 1, dtype=float)
            mag2 = idx ** (-q) - (idx + 1) ** (-q)
            mag2 = mag2 / mag2.sum() * (1 - (N + 1.0) ** (-q))
            planted_theta = rng.choice([-1.0, 1.0], N) * np.sqrt(mag2)
    inst = make_instance(cfg["K"], cfg["Kp"], seed=cfg["inst_seed"],
                         cost_kind=cfg.get("cost_kind", "smooth"),
                         planted_theta=planted_theta)
    Cbar = cfg.get("Cbar_mult", 1.1) * np.linalg.norm(inst.theta_star)

    e = cfg["eps"]
    if e["kind"] == "const":
        eps_schedule = lambda t, v=e["v"]: v
    else:
        eps_schedule = lambda t, eta=e["eta"]: eta * t ** (-eta)

    n_schedule = None
    ns = cfg.get("n_sched")
    if ns:
        if ns["kind"] == "const":
            n_schedule = lambda t, v=ns["v"]: v
        else:
            n_schedule = lambda t, q=ns["q"]: int(np.ceil(t ** (1.0 / (q + 1.0))))

    return run_entucb(inst, T=cfg["T"], sigma=cfg["sigma"], delta=cfg["delta"],
                      lam=cfg["lam"], Cbar=Cbar, eps_schedule=eps_schedule,
                      n_schedule=n_schedule, seed=cfg["run_seed"],
                      opt_iters=cfg.get("opt_iters", 3),
                      kappa=cfg.get("kappa", 0.0), eta=cfg.get("eta_for_bound"),
                      beta_scale=cfg.get("beta_scale", 1.0),
                      sigma_run=cfg.get("sigma_run"))


def parallel_runs(cfgs: list[dict], procs: int = 0) -> list[RunResult]:
    import os
    from concurrent.futures import ProcessPoolExecutor
    procs = procs or min(len(cfgs), os.cpu_count() or 1)
    if procs <= 1 or len(cfgs) == 1:
        return [run_from_config(c) for c in cfgs]
    with ProcessPoolExecutor(max_workers=procs) as ex:
        return list(ex.map(run_from_config, cfgs))