feat: add MACD, Bollinger, VWAP, ATR, EMA, SMA-200 indicator computations

tests/test_indicators.py

@@ -0,0 +1,263 @@
+"""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