39 lines
1.5 KiB
Python
39 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
import numpy as np
|
|
import pytest
|
|
|
|
from control_plane.trading_studio.indicators.historical_band_channel import (
|
|
NATIVE_EVALUATORS,
|
|
OBSERVED_BAND_CHANNEL_PARAMS,
|
|
evaluate_band_channel,
|
|
)
|
|
from control_plane.trading_studio.indicators.historical_formulae import compute_indicator
|
|
|
|
|
|
def _ohlcv() -> tuple[np.ndarray, ...]:
|
|
close = np.linspace(100.0, 140.0, 80) + np.sin(np.arange(80) / 3)
|
|
high = close + 1.5 + (np.arange(80) % 3) / 10
|
|
low = close - 1.25 - (np.arange(80) % 4) / 10
|
|
volume = np.linspace(1_000.0, 2_000.0, 80)
|
|
return close, high, low, volume
|
|
|
|
|
|
def test_batch01_registers_only_the_97_observed_band_channel_primitives():
|
|
assert set(NATIVE_EVALUATORS) == {17, 18, 19, 20, 21, 23, 24, 28}
|
|
assert len(OBSERVED_BAND_CHANNEL_PARAMS) == 97
|
|
assert (22, 10, 0.0) not in OBSERVED_BAND_CHANNEL_PARAMS
|
|
|
|
|
|
@pytest.mark.parametrize("indicator_id,period,p1", sorted(OBSERVED_BAND_CHANNEL_PARAMS))
|
|
def test_batch01_matches_recovered_historical_formulae(indicator_id: int, period: int, p1: float):
|
|
close, high, low, volume = _ohlcv()
|
|
actual = evaluate_band_channel(indicator_id, close, high, low, volume, period, p1)
|
|
expected = compute_indicator(indicator_id, close, high, low, volume, period, p1)
|
|
np.testing.assert_array_equal(actual, expected)
|
|
|
|
|
|
def test_batch01_refuses_unobserved_parameter_expansion():
|
|
close, high, low, volume = _ohlcv()
|
|
with pytest.raises(ValueError, match="unsupported Batch01"):
|
|
evaluate_band_channel(20, close, high, low, volume, 12, 0.0)
|