Cerbero-mcp/services/common/tests/test_indicators.py

import math

from mcp_common.indicators import (
    adx,
    atr,
    autocorrelation,
    garch11_forecast,
    half_life_mean_reversion,
    hurst_exponent,
    macd,
    rolling_sharpe,
    rsi,
    sma,
    var_cvar,
    vol_cone,
)


def test_rsi_simple():
    closes = [44, 44.34, 44.09, 44.15, 43.61, 44.33, 44.83, 45.10, 45.42, 45.84,
              46.08, 45.89, 46.03, 45.61, 46.28]
    r = rsi(closes, period=14)
    assert r is not None
    # Known textbook RSI value ballpark
    assert 65.0 < r < 75.0


def test_rsi_insufficient_data():
    assert rsi([1, 2, 3], period=14) is None


def test_sma_simple():
    assert sma([1, 2, 3, 4, 5], period=5) == 3.0
    assert sma([1, 2, 3], period=5) is None


def test_atr_simple():
    # highs, lows, closes
    highs =  [10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23, 24]
    lows =   [ 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22, 23]
    closes = [9.5,10.5,11.5,12.5,13.5,14.5,15.5,16.5,17.5,18.5,19.5,20.5,21.5,22.5,23.5]
    a = atr(highs, lows, closes, period=14)
    assert a is not None
    assert 0.9 < a <= 1.5


def test_macd_trend_up():
    # monotonic uptrend → MACD > 0, histogram > 0
    closes = [float(i) for i in range(1, 60)]
    m = macd(closes)
    assert m["macd"] is not None
    assert m["signal"] is not None
    assert m["hist"] is not None
    assert m["macd"] > 0
    assert m["hist"] >= 0


def test_macd_insufficient_data():
    m = macd([1.0, 2.0, 3.0])
    assert m == {"macd": None, "signal": None, "hist": None}


def test_macd_trend_down():
    closes = [float(i) for i in range(60, 1, -1)]
    m = macd(closes)
    assert m["macd"] < 0
    assert m["hist"] <= 0


def test_adx_insufficient_data():
    a = adx([1.0] * 10, [0.5] * 10, [0.7] * 10, period=14)
    assert a == {"adx": None, "+di": None, "-di": None}


def test_adx_strong_uptrend():
    highs = [float(i) + 1.0 for i in range(1, 40)]
    lows = [float(i) for i in range(1, 40)]
    closes = [float(i) + 0.5 for i in range(1, 40)]
    a = adx(highs, lows, closes, period=14)
    assert a["adx"] is not None
    assert a["+di"] is not None and a["-di"] is not None
    # strong uptrend → +DI >> -DI, ADX high
    assert a["+di"] > a["-di"]
    assert a["adx"] > 50.0


def test_adx_flat_market():
    highs = [10.0] * 40
    lows = [9.0] * 40
    closes = [9.5] * 40
    a = adx(highs, lows, closes, period=14)
    # no directional movement → ADX near 0
    assert a["adx"] is not None
    assert a["adx"] < 5.0


# ---------- vol_cone ----------

def _gbm_series(mu: float, sigma: float, n: int, seed: int = 42) -> list[float]:
    """Mock GBM closes: deterministic for tests."""
    import random
    r = random.Random(seed)
    p = [100.0]
    for _ in range(n):
        z = r.gauss(0.0, 1.0)
        p.append(p[-1] * math.exp(mu / 252 + sigma / math.sqrt(252) * z))
    return p


def test_vol_cone_returns_percentiles_per_window():
    closes = _gbm_series(mu=0.0, sigma=0.5, n=400)
    out = vol_cone(closes, windows=[10, 30, 60])
    assert set(out.keys()) == {10, 30, 60}
    for _w, stats in out.items():
        assert "current" in stats
        assert "p10" in stats and "p50" in stats and "p90" in stats
        assert stats["p10"] <= stats["p50"] <= stats["p90"]
        # annualized — sensible range for sigma=0.5
        assert 0.1 < stats["p50"] < 1.5


def test_vol_cone_insufficient_data():
    out = vol_cone([100.0, 101.0], windows=[10, 30])
    assert out[10]["current"] is None
    assert out[30]["current"] is None


# ---------- hurst_exponent ----------

def test_hurst_random_walk_near_half():
    closes = _gbm_series(mu=0.0, sigma=0.3, n=500, seed=7)
    h = hurst_exponent(closes)
    assert h is not None
    # Random walk → Hurst ≈ 0.5; R/S bias positivo ben noto su sample finiti.
    # Bound largo: distinguere comunque random walk da trending forte (>0.85).
    assert 0.35 < h < 0.85


def test_hurst_persistent_trend():
    # Strong monotonic trend → H >> 0.5
    closes = [100.0 + i * 0.5 + math.sin(i / 10) * 0.1 for i in range(400)]
    h = hurst_exponent(closes)
    assert h is not None
    assert h > 0.85


def test_hurst_insufficient_data():
    assert hurst_exponent([1.0, 2.0, 3.0]) is None


# ---------- half_life_mean_reversion ----------

def test_half_life_mean_reverting_series():
    """OU process with theta=0.1 → half-life ≈ ln(2)/0.1 ≈ 6.93."""
    import random
    r = random.Random(123)
    theta = 0.1
    mu = 100.0
    sigma = 0.5
    s = [mu]
    for _ in range(500):
        s.append(s[-1] + theta * (mu - s[-1]) + sigma * r.gauss(0, 1))
    hl = half_life_mean_reversion(s)
    assert hl is not None
    # broad tolerance — finite-sample noise
    assert 3.0 < hl < 20.0


def test_half_life_trending_returns_none():
    closes = [100.0 + i for i in range(200)]
    hl = half_life_mean_reversion(closes)
    # No mean reversion → returns None or +inf
    assert hl is None or hl > 1000


# ---------- garch11_forecast ----------

def test_garch11_forecast_returns_positive_sigma():
    closes = _gbm_series(mu=0.0, sigma=0.4, n=500, seed=11)
    out = garch11_forecast(closes)
    assert out is not None
    assert out["sigma_next"] > 0
    assert 0 < out["alpha"] < 1
    assert 0 < out["beta"] < 1
    assert out["alpha"] + out["beta"] < 1.0  # stationarity


def test_garch11_insufficient_data():
    assert garch11_forecast([100.0, 101.0]) is None


# ---------- autocorrelation ----------

def test_autocorrelation_white_noise_low():
    import random
    r = random.Random(1)
    rets = [r.gauss(0, 0.01) for _ in range(500)]
    out = autocorrelation(rets, max_lag=5)
    assert len(out) == 5
    # white noise → all autocorr ≈ 0 (within ±2/sqrt(N))
    bound = 2.0 / math.sqrt(len(rets))
    for _lag, val in out.items():
        assert abs(val) < bound * 2  # generous


def test_autocorrelation_lag1_strong_for_ar1():
    """AR(1) with phi=0.7 → autocorr lag-1 ≈ 0.7."""
    import random
    r = random.Random(2)
    s = [0.0]
    for _ in range(500):
        s.append(0.7 * s[-1] + r.gauss(0, 0.1))
    out = autocorrelation(s, max_lag=3)
    assert out[1] > 0.5
    assert out[2] > 0.2  # geometric decay


def test_autocorrelation_insufficient_data():
    assert autocorrelation([1.0], max_lag=5) == {}


# ---------- rolling_sharpe ----------

def test_rolling_sharpe_positive_for_uptrend():
    closes = [100.0 * (1 + 0.001 * i) for i in range(252)]
    s = rolling_sharpe(closes, window=60)
    assert s is not None
    assert s["sharpe"] > 0
    assert s["sortino"] >= s["sharpe"] / 2  # sortino can be high if no downside


def test_rolling_sharpe_zero_volatility():
    closes = [100.0] * 100
    s = rolling_sharpe(closes, window=60)
    assert s is not None
    assert s["sharpe"] == 0.0  # no variance → 0 by convention


def test_rolling_sharpe_insufficient_data():
    assert rolling_sharpe([100.0, 101.0], window=60) is None


# ---------- var_cvar ----------

def test_var_cvar_basic():
    import random
    r = random.Random(3)
    rets = [r.gauss(0.0005, 0.02) for _ in range(1000)]
    out = var_cvar(rets, confidences=[0.95, 0.99])
    assert "var_95" in out and "cvar_95" in out
    assert "var_99" in out and "cvar_99" in out
    # VaR is loss → positive number representing percentile loss
    assert out["var_95"] > 0
    assert out["cvar_95"] >= out["var_95"]  # CVaR worse than VaR
    assert out["var_99"] >= out["var_95"]


def test_var_cvar_insufficient_data():
    assert var_cvar([0.01], confidences=[0.95]) == {}