dennisthiessen
dc08a805a8
Replace inert ±0.25 bands with ~p25/p75 cutoffs from the prod snapshot (−0.08 / 0.00). Document zero-return dilution and left-skewed distribution.
315 lines
12 KiB
Python
315 lines
12 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_fip_id,
|
||
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)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# FIP ID (display / research context — not a gate)
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class TestComputeFipId:
|
||
def test_steady_climber_is_continuous(self):
|
||
# Many small up days → low (negative) ID for a positive-return path.
|
||
closes = [100.0 * (1.002 ** i) for i in range(280)]
|
||
result = compute_fip_id(closes)
|
||
assert result["fip_id"] < 0
|
||
assert result["path"] == "continuous"
|
||
assert result["display_only"] is True
|
||
assert 0 <= result["score"] <= 100
|
||
|
||
def test_jump_then_flat_is_more_discrete_than_steady(self):
|
||
# Ordering only: zeros dilute %pos/%neg so a pure jump can land near
|
||
# zero ("mixed") rather than "discrete" — paper-faithful, not a bug.
|
||
steady = [100.0 * (1.002 ** i) for i in range(280)]
|
||
jumpy = [100.0] * 252
|
||
jumpy.append(100.0 * 1.5)
|
||
jumpy.extend([100.0 * 1.5] * 40)
|
||
id_steady = compute_fip_id(steady)["fip_id"]
|
||
id_jumpy = compute_fip_id(jumpy)["fip_id"]
|
||
assert id_jumpy > id_steady
|
||
|
||
def test_label_thresholds_match_live_equity_scale(self):
|
||
# Calibrated cutoffs: continuous ≤ −0.08, discrete ≥ 0 (not ±0.25).
|
||
from app.services.indicator_service import (
|
||
FIP_PATH_CONTINUOUS_MAX,
|
||
FIP_PATH_DISCRETE_MIN,
|
||
)
|
||
|
||
assert FIP_PATH_CONTINUOUS_MAX == pytest.approx(-0.08)
|
||
assert FIP_PATH_DISCRETE_MIN == pytest.approx(0.0)
|
||
# Mild continuous (typical momentum winner scale).
|
||
mild = [100.0]
|
||
for i in range(1, 280):
|
||
# ~54% up days with small steps → fip roughly −0.08…−0.12 region.
|
||
mild.append(mild[-1] * (1.0015 if i % 2 == 0 or i % 5 != 0 else 0.9995))
|
||
r = compute_fip_id(mild)
|
||
assert r["fip_id"] > -0.25 # not textbook extreme
|
||
# Labels must fire somewhere in the observed equity band.
|
||
assert r["path"] in ("continuous", "mixed", "discrete")
|
||
|
||
def test_insufficient_data_raises(self):
|
||
with pytest.raises(ValidationError, match="FIP ID requires"):
|
||
compute_fip_id(_rising_closes(50))
|