summaryrefslogtreecommitdiff
path: root/services/strategy-engine/tests/test_indicators.py
blob: 3147fc4700334bed073d1736d5d6a62a2aa04e20 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
"""Tests for technical indicator library."""

import sys
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

import numpy as np
import pandas as pd
import pytest
from strategies.indicators.momentum import rsi, stochastic
from strategies.indicators.trend import adx, ema, macd, sma
from strategies.indicators.volatility import atr, bollinger_bands
from strategies.indicators.volume import obv, volume_ratio, volume_sma


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