feat(strategy): add technical indicators library (ATR, ADX, RSI, MACD, Bollinger, Stochastic, OBV)

1 files changed, 115 insertions, 0 deletions

diff --git a/services/strategy-engine/tests/test_indicators.py b/services/strategy-engine/tests/test_indicators.py
new file mode 100644
index 0000000..ac5b505
--- /dev/null
+++ b/services/strategy-engine/tests/test_indicators.py
@@ -0,0 +1,115 @@
+"""Tests for technical indicator library."""
+import sys
+from pathlib import Path
+sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
+
+import pandas as pd
+import numpy as np
+import pytest
+
+from strategies.indicators.trend import sma, ema, macd, adx
+from strategies.indicators.volatility import atr, bollinger_bands
+from strategies.indicators.momentum import rsi, stochastic
+from strategies.indicators.volume import volume_sma, volume_ratio, obv
+
+
+class TestTrend:
+    def test_sma_basic(self):
+        s = pd.Series([1, 2, 3, 4, 5])
+        result = sma(s, 3)
+        assert result.iloc[-1] == pytest.approx(4.0)
+
+    def test_ema_basic(self):
+        s = pd.Series([1, 2, 3, 4, 5])
+        result = ema(s, 3)
+        assert not np.isnan(result.iloc[-1])
+
+    def test_macd_returns_three_series(self):
+        s = pd.Series(range(50), dtype=float)
+        macd_line, signal_line, hist = macd(s)
+        assert len(macd_line) == 50
+        assert len(signal_line) == 50
+        assert len(hist) == 50
+
+    def test_adx_returns_series(self):
+        n = 60
+        highs = pd.Series(np.random.uniform(100, 110, n))
+        lows = pd.Series(np.random.uniform(90, 100, n))
+        closes = pd.Series(np.random.uniform(95, 105, n))
+        result = adx(highs, lows, closes, period=14)
+        assert len(result) == n
+        # Last value should be a number (not NaN for 60 bars with period 14)
+        assert not np.isnan(result.iloc[-1])
+
+    def test_adx_trending_market(self):
+        # Strong uptrend should have high ADX
+        n = 100
+        closes = pd.Series([100 + i * 2 for i in range(n)], dtype=float)
+        highs = closes + 3
+        lows = closes - 3
+        result = adx(highs, lows, closes)
+        assert result.iloc[-1] > 25  # Strong trend
+
+    def test_adx_ranging_market(self):
+        # Sideways market should have low ADX
+        n = 100
+        closes = pd.Series([100 + (i % 5) - 2 for i in range(n)], dtype=float)
+        highs = closes + 1
+        lows = closes - 1
+        result = adx(highs, lows, closes)
+        assert result.iloc[-1] < 25
+
+
+class TestVolatility:
+    def test_atr_basic(self):
+        n = 30
+        highs = pd.Series([110] * n, dtype=float)
+        lows = pd.Series([90] * n, dtype=float)
+        closes = pd.Series([100] * n, dtype=float)
+        result = atr(highs, lows, closes, period=14)
+        assert result.iloc[-1] == pytest.approx(20.0, rel=0.01)
+
+    def test_bollinger_bands_width(self):
+        s = pd.Series([100] * 20 + [110, 90] * 5, dtype=float)
+        upper, mid, lower = bollinger_bands(s, period=20)
+        assert upper.iloc[-1] > mid.iloc[-1] > lower.iloc[-1]
+
+
+class TestMomentum:
+    def test_rsi_oversold(self):
+        # Declining prices should give low RSI
+        s = pd.Series([100 - i for i in range(30)], dtype=float)
+        result = rsi(s, period=14)
+        assert result.iloc[-1] < 30
+
+    def test_rsi_overbought(self):
+        s = pd.Series([100 + i for i in range(30)], dtype=float)
+        result = rsi(s, period=14)
+        assert result.iloc[-1] > 70
+
+    def test_stochastic_returns_k_and_d(self):
+        n = 30
+        highs = pd.Series(np.random.uniform(100, 110, n))
+        lows = pd.Series(np.random.uniform(90, 100, n))
+        closes = pd.Series(np.random.uniform(95, 105, n))
+        k, d = stochastic(highs, lows, closes)
+        assert len(k) == n
+        assert len(d) == n
+
+
+class TestVolume:
+    def test_volume_sma(self):
+        v = pd.Series([100] * 20 + [200] * 5, dtype=float)
+        result = volume_sma(v, period=20)
+        assert result.iloc[-1] > 100
+
+    def test_volume_ratio_above_average(self):
+        v = pd.Series([100] * 20 + [300], dtype=float)
+        result = volume_ratio(v, period=20)
+        assert result.iloc[-1] > 2.0
+
+    def test_obv_up(self):
+        closes = pd.Series([100, 101, 102, 103], dtype=float)
+        volumes = pd.Series([10, 10, 10, 10], dtype=float)
+        result = obv(closes, volumes)
+        assert result.iloc[-1] > 0  # All up moves = positive OBV