"""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