Adaptive-Barrier-Monitor/tests/test_engine.py

import numpy as np
import pytest

from adaptive_barrier.engine import (
    TRADING_MINUTES_PER_YEAR,
    adaptive_schedule,
    barrier_miss_prob,
    detect_barrier_event,
    fixed_uniform_indices,
    geometric_brownian_motion,
    horizon_vol,
    max_safe_dt,
    merton_jump_diffusion,
    run_monte_carlo_simulation,
)


def test_symmetric_bridge_inversion_matches_epsilon():
    distance = 0.04
    sigma = 0.01
    epsilon = 1e-3
    dt = max_safe_dt(distance, sigma, epsilon)
    assert barrier_miss_prob(distance, distance, 0.0, sigma, dt) == pytest.approx(epsilon)


def test_vectorized_safe_interval_and_validation():
    result = max_safe_dt(np.array([0.01, 0.02]), 0.1, 1e-3)
    assert result[1] == pytest.approx(4.0 * result[0])
    with pytest.raises(ValueError):
        max_safe_dt(-0.01, 0.1, 1e-3)


def test_adaptive_schedule_uses_matching_minute_units():
    times = np.linspace(0.0, 10.0, 101)
    path = np.zeros_like(times)
    sigma_per_sqrt_minute = horizon_vol(0.30, 1.0)
    indices = adaptive_schedule(
        times,
        path,
        np.log(0.9),
        sigma_per_sqrt_minute,
        1e-3,
        dt_cap=2.0,
    )
    assert np.array_equal(indices, np.array([0, 20, 40, 60, 80, 100]))


def test_detection_deadline_is_enforced():
    values = np.array([100.0, 89.0, 88.0, 87.0, 86.0])
    samples = np.array([0, 4])
    assert detect_barrier_event(samples, values, 1, 90.0, "down", 2) == (False, None, None)
    assert detect_barrier_event(samples, values, 1, 90.0, "down", 3) == (True, 3, 4)


def test_detection_rejects_invalid_indices():
    values = np.array([100.0, 89.0, 88.0])
    with pytest.raises(ValueError, match="within the path"):
        detect_barrier_event(np.array([0, 3]), values, 1, 90.0, "down", 2)
    with pytest.raises(ValueError, match="ascending"):
        detect_barrier_event(np.array([2, 1]), values, 1, 90.0, "down", 2)
    with pytest.raises(ValueError, match="integer grid indices"):
        detect_barrier_event(np.array([0.5, 1.0]), values, 1, 90.0, "down", 2)


def test_fixed_indices_have_exact_requested_count():
    for n in (2, 10, 101):
        for k in range(1, n + 1):
            indices = fixed_uniform_indices(n, k)
            assert len(indices) == k
            assert len(np.unique(indices)) == k
            assert indices[0] == 0
            if k > 1:
                assert indices[-1] == n - 1


def test_zero_jump_intensity_matches_gbm_for_same_seed():
    kwargs = dict(S0=100.0, T=5 / TRADING_MINUTES_PER_YEAR, n_steps=100, n_paths=3, mu=0.05, sigma=0.2)
    _, gbm = geometric_brownian_motion(**kwargs, rng=np.random.default_rng(9))
    _, jump = merton_jump_diffusion(
        **kwargs,
        jump_intensity=0.0,
        rng=np.random.default_rng(9),
    )
    assert np.array_equal(gbm, jump)


def test_merton_jump_diffusion_validates_time_grid_before_sampling():
    with pytest.raises(ValueError, match="T"):
        merton_jump_diffusion(100.0, 0.0, 100)
    with pytest.raises(ValueError, match="n_steps"):
        merton_jump_diffusion(100.0, 5 / TRADING_MINUTES_PER_YEAR, 0)


def test_simulation_baseline_has_exact_per_path_budget():
    result = run_monte_carlo_simulation(
        n_paths=12,
        n_steps=200,
        rng_seed=7,
        drop_fraction=0.05,
        rise_fraction=0.08,
    )
    assert result["adaptive_total_samples"] == result["fixed_total_samples"]
    for path in result["paths"]:
        assert len(path["sample_indices"]) == len(path["fixed_sample_indices"])


def test_simulation_lags_respect_deadline():
    deadline = 3
    result = run_monte_carlo_simulation(
        n_paths=20,
        n_steps=300,
        rng_seed=3,
        drop_fraction=0.04,
        rise_fraction=0.06,
        max_detection_lag_steps=deadline,
    )
    for path in result["paths"]:
        for key in (
            "adaptive_lower_detection_lag",
            "adaptive_upper_detection_lag",
            "fixed_lower_detection_lag",
            "fixed_upper_detection_lag",
        ):
            lag = path[key]
            assert lag is None or 0 <= lag <= deadline


def test_simulation_exposes_unambiguous_event_counts():
    result = run_monte_carlo_simulation(
        n_paths=8,
        n_steps=150,
        rng_seed=11,
        drop_fraction=0.04,
        rise_fraction=0.06,
    )
    assert result["n_barrier_events"] == result["n_lower_events"] + result["n_upper_events"]
    assert result["n_breaches"] == result["n_barrier_events"]
    assert 0 <= result["n_paths_with_any_breach"] <= result["n_paths"]


def test_fixed_cadence_indices_include_window_end_and_respect_grid():
    from adaptive_barrier.engine import fixed_cadence_indices

    times = np.linspace(0.0, 100.0, 11)
    indices = fixed_cadence_indices(times, 25.0)
    assert np.array_equal(indices, np.array([0, 3, 5, 8, 10]))


def test_fixed_cadence_mode_uses_independent_sample_count():
    result = run_monte_carlo_simulation(
        n_paths=4,
        n_steps=200,
        window_minutes=200.0,
        dt_cap_minutes=2.0,
        comparison_mode="fixed_cadence",
        fixed_cadence_minutes=20.0,
        rng_seed=5,
    )
    assert result["comparison_mode"] == "fixed_cadence"
    assert result["fixed_cadence_minutes"] == 20.0
    assert result["adaptive_total_samples"] != result["fixed_total_samples"]
    fixed_counts = {len(path["fixed_sample_indices"]) for path in result["paths"]}
    assert fixed_counts == {11}