265 lines
10 KiB
Python
265 lines
10 KiB
Python
"""Unit tests for app.services.indicator_service pure computation functions."""
|
|
|
|
import pytest
|
|
|
|
from app.exceptions import ValidationError
|
|
from app.services.indicator_service import (
|
|
compute_adx,
|
|
compute_atr,
|
|
compute_ema,
|
|
compute_ema_cross,
|
|
compute_pivot_points,
|
|
compute_rsi,
|
|
compute_volume_profile,
|
|
)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers: generate synthetic OHLCV data
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _rising_closes(n: int, start: float = 100.0, step: float = 1.0) -> list[float]:
|
|
return [start + i * step for i in range(n)]
|
|
|
|
|
|
def _flat_closes(n: int, price: float = 100.0) -> list[float]:
|
|
return [price] * n
|
|
|
|
|
|
def _ohlcv_from_closes(closes: list[float], spread: float = 2.0):
|
|
"""Generate highs/lows/volumes from a close series."""
|
|
highs = [c + spread for c in closes]
|
|
lows = [c - spread for c in closes]
|
|
volumes = [1000] * len(closes)
|
|
return highs, lows, closes, volumes
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# EMA
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestComputeEMA:
|
|
def test_basic_ema(self):
|
|
closes = _rising_closes(25)
|
|
result = compute_ema(closes, period=20)
|
|
assert "ema" in result
|
|
assert "score" in result
|
|
assert 0 <= result["score"] <= 100
|
|
|
|
def test_insufficient_data_raises(self):
|
|
closes = _rising_closes(5)
|
|
with pytest.raises(ValidationError, match="EMA.*requires at least"):
|
|
compute_ema(closes, period=20)
|
|
|
|
def test_price_above_ema_high_score(self):
|
|
# Rising prices → latest close above EMA → score > 50
|
|
closes = _rising_closes(30, start=100, step=2)
|
|
result = compute_ema(closes, period=20)
|
|
assert result["score"] > 50
|
|
|
|
def test_price_below_ema_low_score(self):
|
|
# Falling prices → latest close below EMA → score < 50
|
|
closes = list(reversed(_rising_closes(30, start=100, step=2)))
|
|
result = compute_ema(closes, period=20)
|
|
assert result["score"] < 50
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# RSI
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestComputeRSI:
|
|
def test_basic_rsi(self):
|
|
closes = _rising_closes(20)
|
|
result = compute_rsi(closes)
|
|
assert "rsi" in result
|
|
assert 0 <= result["score"] <= 100
|
|
|
|
def test_all_gains_rsi_100(self):
|
|
closes = _rising_closes(20, step=1)
|
|
result = compute_rsi(closes)
|
|
assert result["rsi"] == 100.0
|
|
|
|
def test_all_losses_rsi_0(self):
|
|
closes = list(reversed(_rising_closes(20, step=1)))
|
|
result = compute_rsi(closes)
|
|
assert result["rsi"] == pytest.approx(0.0, abs=0.5)
|
|
|
|
def test_insufficient_data_raises(self):
|
|
with pytest.raises(ValidationError, match="RSI requires"):
|
|
compute_rsi([100.0] * 5)
|
|
|
|
def test_overbought_rsi_is_penalized_not_maximal(self):
|
|
"""RSI 100 (extreme overbought) must NOT score near 100."""
|
|
from app.services.indicator_service import _rsi_to_score
|
|
|
|
assert _rsi_to_score(100.0) < 40.0 # overbought penalized
|
|
assert _rsi_to_score(90.0) < _rsi_to_score(60.0) # extreme < healthy
|
|
assert _rsi_to_score(60.0) > 80.0 # healthy momentum rewarded
|
|
# All gains → RSI 100 → low score, not 100
|
|
result = compute_rsi(_rising_closes(20, step=1))
|
|
assert result["score"] < 40.0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ATR
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestComputeATR:
|
|
def test_basic_atr(self):
|
|
closes = _rising_closes(20)
|
|
highs, lows, _, _ = _ohlcv_from_closes(closes)
|
|
result = compute_atr(highs, lows, closes)
|
|
assert "atr" in result
|
|
assert result["atr"] > 0
|
|
assert 0 <= result["score"] <= 100
|
|
|
|
def test_insufficient_data_raises(self):
|
|
closes = [100.0] * 5
|
|
highs, lows, _, _ = _ohlcv_from_closes(closes)
|
|
with pytest.raises(ValidationError, match="ATR requires"):
|
|
compute_atr(highs, lows, closes)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ADX
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestComputeADX:
|
|
def test_basic_adx(self):
|
|
closes = _rising_closes(30)
|
|
highs, lows, _, _ = _ohlcv_from_closes(closes)
|
|
result = compute_adx(highs, lows, closes)
|
|
assert "adx" in result
|
|
assert "plus_di" in result
|
|
assert "minus_di" in result
|
|
assert 0 <= result["score"] <= 100
|
|
|
|
def test_insufficient_data_raises(self):
|
|
closes = _rising_closes(10)
|
|
highs, lows, _, _ = _ohlcv_from_closes(closes)
|
|
with pytest.raises(ValidationError, match="ADX requires"):
|
|
compute_adx(highs, lows, closes)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Volume Profile
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestComputeVolumeProfile:
|
|
def test_basic_volume_profile(self):
|
|
closes = _rising_closes(25)
|
|
highs, lows, _, volumes = _ohlcv_from_closes(closes)
|
|
result = compute_volume_profile(highs, lows, closes, volumes)
|
|
assert "poc" in result
|
|
assert "value_area_low" in result
|
|
assert "value_area_high" in result
|
|
assert "hvn" in result
|
|
assert "lvn" in result
|
|
assert 0 <= result["score"] <= 100
|
|
|
|
def test_insufficient_data_raises(self):
|
|
closes = [100.0] * 10
|
|
highs, lows, _, volumes = _ohlcv_from_closes(closes)
|
|
with pytest.raises(ValidationError, match="Volume Profile requires"):
|
|
compute_volume_profile(highs, lows, closes, volumes)
|
|
|
|
def test_close_bin_volume_no_double_count(self):
|
|
"""Each bar's volume is counted once (close bin), not per span."""
|
|
# Wide bars that would span many bins under the old algorithm
|
|
n = 25
|
|
closes = [100.0 + (i % 5) for i in range(n)]
|
|
highs = [c + 20 for c in closes] # wide range
|
|
lows = [c - 20 for c in closes]
|
|
volumes = [1000] * n
|
|
result = compute_volume_profile(highs, lows, closes, volumes, num_bins=20)
|
|
# Binned total equals true volume (close-bin assignment)
|
|
# We only expose poc/hvn; reconstruct by checking score fields exist
|
|
assert result["poc"] > 0
|
|
# With volume concentrated on a few close prices, HVNs should be few local peaks
|
|
assert len(result["hvn"]) < 20
|
|
|
|
def test_hvn_are_local_peaks_not_all_above_mean(self):
|
|
"""HVN should be local histogram peaks, not every above-mean bin."""
|
|
# Two clusters of closes → two volume peaks
|
|
closes = [80.0] * 10 + [120.0] * 10 + [100.0] * 5
|
|
highs = [c + 1 for c in closes]
|
|
lows = [c - 1 for c in closes]
|
|
volumes = [1000] * len(closes)
|
|
result = compute_volume_profile(highs, lows, closes, volumes, num_bins=20)
|
|
# At most a handful of local peaks (not ~half of 20 bins)
|
|
assert len(result["hvn"]) <= 6
|
|
# POC should land near one of the high-volume clusters
|
|
assert result["poc"] < 95 or result["poc"] > 105
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Pivot Points
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestComputePivotPoints:
|
|
def test_basic_pivot_points(self):
|
|
# Create data with clear swing highs/lows
|
|
closes = [10, 15, 20, 15, 10, 15, 20, 15, 10, 15]
|
|
highs = [c + 1 for c in closes]
|
|
lows = [c - 1 for c in closes]
|
|
result = compute_pivot_points(highs, lows, closes)
|
|
assert "swing_highs" in result
|
|
assert "swing_lows" in result
|
|
assert 0 <= result["score"] <= 100
|
|
|
|
def test_insufficient_data_raises(self):
|
|
with pytest.raises(ValidationError, match="Pivot Points requires"):
|
|
compute_pivot_points([1, 2], [0, 1], [0.5, 1.5])
|
|
|
|
def test_prominence_filters_tiny_swings(self):
|
|
# Mix of a large swing (depth ~10) and tiny fractal noise (depth ~1)
|
|
closes = [
|
|
10, 10.2, 10.5, 10.2, 10, # tiny high around idx 2
|
|
10, 15, 20, 15, 10, # large high around idx 7
|
|
10, 10.3, 10.6, 10.3, 10, # tiny high around idx 12
|
|
]
|
|
highs = list(closes)
|
|
lows = [c - 0.5 for c in closes]
|
|
highs[2] = 10.8
|
|
highs[7] = 20.5
|
|
highs[12] = 10.9
|
|
lows[7] = 10.0 # large window range at major swing
|
|
unfiltered = compute_pivot_points(highs, lows, closes, min_prominence=None)
|
|
filtered = compute_pivot_points(highs, lows, closes, min_prominence=5.0)
|
|
assert unfiltered["pivot_count"] > 0
|
|
assert filtered["pivot_count"] < unfiltered["pivot_count"]
|
|
# Major swing high should survive
|
|
assert any(h >= 20.0 for h in filtered["swing_highs"])
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# EMA Cross
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestComputeEMACross:
|
|
def test_bullish_signal(self):
|
|
# Rising prices → short EMA > long EMA → bullish
|
|
closes = _rising_closes(60, step=2)
|
|
result = compute_ema_cross(closes, short_period=20, long_period=50)
|
|
assert result["signal"] == "bullish"
|
|
assert result["short_ema"] > result["long_ema"]
|
|
|
|
def test_bearish_signal(self):
|
|
# Falling prices → short EMA < long EMA → bearish
|
|
closes = list(reversed(_rising_closes(60, step=2)))
|
|
result = compute_ema_cross(closes, short_period=20, long_period=50)
|
|
assert result["signal"] == "bearish"
|
|
assert result["short_ema"] < result["long_ema"]
|
|
|
|
def test_neutral_signal(self):
|
|
# Flat prices → EMAs converge → neutral
|
|
closes = _flat_closes(60)
|
|
result = compute_ema_cross(closes, short_period=20, long_period=50)
|
|
assert result["signal"] == "neutral"
|
|
|
|
def test_insufficient_data_raises(self):
|
|
closes = _rising_closes(30)
|
|
with pytest.raises(ValidationError, match="EMA Cross requires"):
|
|
compute_ema_cross(closes, short_period=20, long_period=50)
|