spb/anomaly-atlas Public License

1# =============================================================================2#  Project   : anomaly-atlas3#  File      : benchmarks/synthetic/generators.py4#  Purpose   : Synthetic series with KNOWN properties — the "test the tests" set5#  Author    : Simon-Pierre Boucher6#  Contact   : contact@spboucher.ai7#  Data src  : hfmarketdata.io (sole data source)8#  Created   : 2026-08-129#  Modified  : 2026-08-1210#  Platform  : macOS / Apple Silicon (arm64)11#  License   : All rights reserved (research code)12# =============================================================================13"""Synthetic price/return series with planted, analytically-known properties.1415Charter §8.1: before any detector touches real data it must (a) find NOTHING16in a pure random walk, (b) recover every planted effect, and (c) flag a pure17bid-ask-bounce series as an artifact, not an anomaly. These generators are the18ground truth for that gate (see test_synthetic_gate.py).1920All series are generated from an explicit seed; no global RNG state.21"""2223from __future__ import annotations2425import numpy as np262728def random_walk(n: int, sigma: float = 0.001, seed: int = 0) -> np.ndarray:29    """Pure log-price random walk. Ground truth: VR(q)=1, AC1(returns)=0."""30    rng = np.random.default_rng(seed)31    return np.cumsum(rng.normal(0.0, sigma, n))323334def ou_prices(n: int, kappa: float, sigma: float = 0.001, seed: int = 0) -> np.ndarray:35    """Mean-reverting (Ornstein-Uhlenbeck) log-price around 0.3637    Discrete: p_t = (1 - kappa) * p_{t-1} + eps. Ground truth half-life38    = ln(2) / -ln(1 - kappa); return AC1 < 0; VR(q) < 1 for q >= 2.39    """40    rng = np.random.default_rng(seed)41    p = np.empty(n)42    p[0] = 0.043    eps = rng.normal(0.0, sigma, n)44    for t in range(1, n):45        p[t] = (1.0 - kappa) * p[t - 1] + eps[t]46    return p474849def roll_bounce_prices(n: int, spread: float, sigma: float = 0.001, seed: int = 0) -> np.ndarray:50    """Roll (1984) model: observed log-price = random-walk mid ± spread/2.5152    Ground truth: Cov(r_t, r_{t-1}) = -spread^2/4, implied Roll spread =53    2*sqrt(-cov) = spread, and the WHOLE negative AC1 is artifact.54    """55    rng = np.random.default_rng(seed)56    mid = np.cumsum(rng.normal(0.0, sigma, n))57    q = rng.choice([-1.0, 1.0], size=n)58    return mid + (spread / 2.0) * q596061def leadlag_pair(62    n: int, beta: float, lag: int, sigma: float = 0.001, seed: int = 063) -> tuple[np.ndarray, np.ndarray]:64    """Return series (x, y) where x truly leads y by `lag` steps.6566    y_t = beta * x_{t-lag} + noise. Ground truth: cross-corr peaks at `lag`67    with corr ≈ beta*sd(x)/sd(y); zero at all other lags.68    """69    rng = np.random.default_rng(seed)70    x = rng.normal(0.0, sigma, n)71    noise = rng.normal(0.0, sigma, n)72    y = noise.copy()73    y[lag:] += beta * x[: n - lag]74    return x, y757677def seasonal_returns(78    n: int,79    period: int,80    hot_phase: int,81    amplitude: float,82    sigma: float = 0.001,83    seed: int = 0,84) -> np.ndarray:85    """Returns with a planted calendar effect: mean = amplitude on one phase.8687    Ground truth: mean(returns | t % period == hot_phase) = amplitude,88    all other phases 0.89    """90    rng = np.random.default_rng(seed)91    r = rng.normal(0.0, sigma, n)92    r[np.arange(n) % period == hot_phase] += amplitude93    return r949596def stale_observe(97    prices: np.ndarray, p_observe: float, seed: int = 098) -> tuple[np.ndarray, np.ndarray]:99    """Simulate an illiquid ticker: each price prints with prob p_observe,100    otherwise the last print is carried forward (LOCF).101102    Returns (locf_prices, observed_mask). Ground truth: LOCF returns of a103    random walk gain SPURIOUS positive lag-1 autocorrelation, and a fully104    observed correlated series appears to LEAD the stale one.105    """106    rng = np.random.default_rng(seed)107    observed = rng.random(len(prices)) < p_observe108    observed[0] = True109    locf = prices.copy()110    for t in range(1, len(prices)):111        if not observed[t]:112            locf[t] = locf[t - 1]113    return locf, observed114115116def correlated_pair(117    n: int, rho: float, sigma: float = 0.001, seed: int = 0118) -> tuple[np.ndarray, np.ndarray]:119    """Two random-walk log-prices with contemporaneously correlated innovations.120121    Ground truth: corr(r_x, r_y) = rho at lag 0, zero at every nonzero lag —122    any measured lead-lag after LOCF is pure artifact.123    """124    rng = np.random.default_rng(seed)125    z1 = rng.normal(0.0, sigma, n)126    z2 = rng.normal(0.0, sigma, n)127    rx = z1128    ry = rho * z1 + np.sqrt(1.0 - rho**2) * z2129    return np.cumsum(rx), np.cumsum(ry)130