263 lines
9.3 KiB
Python
263 lines
9.3 KiB
Python
"""Tests for extended technical indicator computations in MarketDataManager."""
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
|
|
from services.signal_generator.market_data import MarketDataManager
|
|
from shared.schemas.trading import OHLCVBar
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
def _bar(
|
|
close: float,
|
|
volume: float = 1000.0,
|
|
high: float | None = None,
|
|
low: float | None = None,
|
|
open_: float | None = None,
|
|
) -> OHLCVBar:
|
|
"""Create a single OHLCVBar with sensible defaults."""
|
|
return OHLCVBar(
|
|
timestamp=datetime.now(tz=timezone.utc),
|
|
open=open_ if open_ is not None else close,
|
|
high=high if high is not None else close,
|
|
low=low if low is not None else close,
|
|
close=close,
|
|
volume=volume,
|
|
)
|
|
|
|
|
|
def _add_bars(
|
|
mgr: MarketDataManager,
|
|
ticker: str,
|
|
closes: list[float],
|
|
volume: float = 1000.0,
|
|
) -> None:
|
|
"""Add multiple bars (one per close) to the manager."""
|
|
for c in closes:
|
|
mgr.add_bar(ticker, _bar(c, volume=volume))
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# EMA
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestEMA:
|
|
"""Tests for exponential moving average computation."""
|
|
|
|
def test_ema_returns_none_insufficient_data(self) -> None:
|
|
mgr = MarketDataManager(max_bars=300)
|
|
_add_bars(mgr, "X", [100.0] * 5)
|
|
snap = mgr.get_snapshot("X")
|
|
assert snap is not None
|
|
assert snap.ema_9 is None
|
|
|
|
def test_ema_9_with_exact_data(self) -> None:
|
|
mgr = MarketDataManager(max_bars=300)
|
|
_add_bars(mgr, "X", [100.0] * 9)
|
|
snap = mgr.get_snapshot("X")
|
|
assert snap is not None
|
|
assert snap.ema_9 == pytest.approx(100.0, abs=0.01)
|
|
|
|
def test_ema_responds_to_recent_prices(self) -> None:
|
|
mgr = MarketDataManager(max_bars=300)
|
|
_add_bars(mgr, "X", [100.0] * 20 + [110.0])
|
|
snap = mgr.get_snapshot("X")
|
|
assert snap is not None
|
|
assert snap.ema_9 is not None
|
|
assert 100.0 < snap.ema_9 < 110.0
|
|
|
|
def test_ema_21_returns_none_insufficient_data(self) -> None:
|
|
mgr = MarketDataManager(max_bars=300)
|
|
_add_bars(mgr, "X", [100.0] * 15)
|
|
snap = mgr.get_snapshot("X")
|
|
assert snap is not None
|
|
assert snap.ema_21 is None
|
|
|
|
def test_ema_21_computed_with_enough_data(self) -> None:
|
|
mgr = MarketDataManager(max_bars=300)
|
|
_add_bars(mgr, "X", [100.0] * 25)
|
|
snap = mgr.get_snapshot("X")
|
|
assert snap is not None
|
|
assert snap.ema_21 == pytest.approx(100.0, abs=0.01)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# SMA-200
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestSMA200:
|
|
"""Tests for 200-period simple moving average."""
|
|
|
|
def test_sma_200_returns_none_insufficient_data(self) -> None:
|
|
mgr = MarketDataManager(max_bars=300)
|
|
_add_bars(mgr, "X", [100.0] * 100)
|
|
snap = mgr.get_snapshot("X")
|
|
assert snap is not None
|
|
assert snap.sma_200 is None
|
|
|
|
def test_sma_200_computed_with_enough_data(self) -> None:
|
|
mgr = MarketDataManager(max_bars=300)
|
|
_add_bars(mgr, "X", [100.0] * 200)
|
|
snap = mgr.get_snapshot("X")
|
|
assert snap is not None
|
|
assert snap.sma_200 == pytest.approx(100.0, abs=0.01)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# MACD
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestMACD:
|
|
"""Tests for MACD(12,26,9) computation."""
|
|
|
|
def test_macd_returns_none_insufficient_data(self) -> None:
|
|
mgr = MarketDataManager(max_bars=300)
|
|
_add_bars(mgr, "X", [100.0] * 20)
|
|
snap = mgr.get_snapshot("X")
|
|
assert snap is not None
|
|
assert snap.macd is None
|
|
|
|
def test_macd_computed_with_enough_data(self) -> None:
|
|
mgr = MarketDataManager(max_bars=300)
|
|
_add_bars(mgr, "X", [100.0] * 40)
|
|
snap = mgr.get_snapshot("X")
|
|
assert snap is not None
|
|
assert snap.macd == pytest.approx(0.0, abs=0.01)
|
|
|
|
def test_macd_positive_in_uptrend(self) -> None:
|
|
mgr = MarketDataManager(max_bars=300)
|
|
rising = [50.0 + i for i in range(50)]
|
|
_add_bars(mgr, "X", rising)
|
|
snap = mgr.get_snapshot("X")
|
|
assert snap is not None
|
|
assert snap.macd is not None
|
|
assert snap.macd > 0
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Bollinger Bands
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestBollingerBands:
|
|
"""Tests for Bollinger Bands (SMA-20 +/- 2 std dev)."""
|
|
|
|
def test_bollinger_returns_none_insufficient_data(self) -> None:
|
|
mgr = MarketDataManager(max_bars=300)
|
|
_add_bars(mgr, "X", [100.0] * 10)
|
|
snap = mgr.get_snapshot("X")
|
|
assert snap is not None
|
|
assert snap.bollinger_upper is None
|
|
|
|
def test_bollinger_computed_with_enough_data(self) -> None:
|
|
mgr = MarketDataManager(max_bars=300)
|
|
_add_bars(mgr, "X", [100.0] * 25)
|
|
snap = mgr.get_snapshot("X")
|
|
assert snap is not None
|
|
assert snap.bollinger_mid == pytest.approx(100.0, abs=0.01)
|
|
assert snap.bollinger_upper is not None
|
|
assert snap.bollinger_lower is not None
|
|
assert snap.bollinger_upper >= snap.bollinger_mid
|
|
assert snap.bollinger_lower <= snap.bollinger_mid
|
|
|
|
def test_bollinger_width_increases_with_volatility(self) -> None:
|
|
# Stable prices -> narrow bands
|
|
mgr_stable = MarketDataManager(max_bars=300)
|
|
_add_bars(mgr_stable, "X", [100.0] * 25)
|
|
snap_stable = mgr_stable.get_snapshot("X")
|
|
|
|
# Alternating prices -> wider bands
|
|
mgr_volatile = MarketDataManager(max_bars=300)
|
|
alternating = [100.0 + (10.0 if i % 2 == 0 else -10.0) for i in range(25)]
|
|
_add_bars(mgr_volatile, "X", alternating)
|
|
snap_volatile = mgr_volatile.get_snapshot("X")
|
|
|
|
assert snap_stable is not None and snap_volatile is not None
|
|
assert snap_stable.bollinger_upper is not None
|
|
assert snap_stable.bollinger_lower is not None
|
|
assert snap_volatile.bollinger_upper is not None
|
|
assert snap_volatile.bollinger_lower is not None
|
|
|
|
width_stable = snap_stable.bollinger_upper - snap_stable.bollinger_lower
|
|
width_volatile = snap_volatile.bollinger_upper - snap_volatile.bollinger_lower
|
|
assert width_volatile > width_stable
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# VWAP
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestVWAP:
|
|
"""Tests for Volume-Weighted Average Price."""
|
|
|
|
def test_vwap_computed(self) -> None:
|
|
mgr = MarketDataManager(max_bars=300)
|
|
for _ in range(5):
|
|
mgr.add_bar("X", _bar(close=100.0, high=100.0, low=100.0, volume=1000.0))
|
|
snap = mgr.get_snapshot("X")
|
|
assert snap is not None
|
|
assert snap.vwap == pytest.approx(100.0, abs=0.01)
|
|
|
|
def test_vwap_weighted_by_volume(self) -> None:
|
|
mgr = MarketDataManager(max_bars=300)
|
|
# Bar 1: typical price = (110+90+100)/3 = 100, volume = 100
|
|
mgr.add_bar("X", _bar(close=100.0, high=110.0, low=90.0, volume=100.0))
|
|
# Bar 2: typical price = (220+180+200)/3 = 200, volume = 900
|
|
mgr.add_bar("X", _bar(close=200.0, high=220.0, low=180.0, volume=900.0))
|
|
snap = mgr.get_snapshot("X")
|
|
assert snap is not None
|
|
assert snap.vwap is not None
|
|
# Expected: (100*100 + 200*900) / (100+900) = 190
|
|
assert snap.vwap == pytest.approx(190.0, abs=0.01)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# ATR
|
|
# ---------------------------------------------------------------------------
|
|
|
|
|
|
class TestATR:
|
|
"""Tests for Average True Range."""
|
|
|
|
def test_atr_returns_none_insufficient_data(self) -> None:
|
|
mgr = MarketDataManager(max_bars=300)
|
|
for _ in range(5):
|
|
mgr.add_bar("X", _bar(close=100.0, high=105.0, low=95.0))
|
|
snap = mgr.get_snapshot("X")
|
|
assert snap is not None
|
|
assert snap.atr is None
|
|
|
|
def test_atr_computed_with_enough_data(self) -> None:
|
|
mgr = MarketDataManager(max_bars=300)
|
|
for _ in range(15):
|
|
mgr.add_bar("X", _bar(close=100.0, high=105.0, low=95.0))
|
|
snap = mgr.get_snapshot("X")
|
|
assert snap is not None
|
|
assert snap.atr is not None
|
|
assert snap.atr >= 0
|
|
|
|
def test_atr_increases_with_volatility(self) -> None:
|
|
# Tight range
|
|
mgr_tight = MarketDataManager(max_bars=300)
|
|
for _ in range(15):
|
|
mgr_tight.add_bar("X", _bar(close=100.0, high=101.0, low=99.0))
|
|
snap_tight = mgr_tight.get_snapshot("X")
|
|
|
|
# Wide range
|
|
mgr_wide = MarketDataManager(max_bars=300)
|
|
for _ in range(15):
|
|
mgr_wide.add_bar("X", _bar(close=100.0, high=120.0, low=80.0))
|
|
snap_wide = mgr_wide.get_snapshot("X")
|
|
|
|
assert snap_tight is not None and snap_wide is not None
|
|
assert snap_tight.atr is not None and snap_wide.atr is not None
|
|
assert snap_wide.atr > snap_tight.atr
|