test_technical_analyzer.pyβ’12.9 kB
"""
Unit tests for TechnicalAnalyzer class
TDD Red Phase: Write failing tests for technical analysis engine
"""
import pytest
import asyncio
from unittest.mock import Mock, AsyncMock, patch
from datetime import datetime, date, timedelta
from typing import Dict, Any, List, Optional
# Test imports - initially will fail (TDD Red phase)
from src.tools.technical_tools import TechnicalAnalyzer
from src.exceptions import MCPStockDetailsError, InsufficientDataError
class TestTechnicalAnalyzer:
"""Test cases for TechnicalAnalyzer class"""
@pytest.fixture
def technical_analyzer(self):
"""Create TechnicalAnalyzer instance for testing"""
return TechnicalAnalyzer()
@pytest.fixture
def sample_ohlcv_data(self):
"""Sample OHLCV data for technical calculations"""
return [
{"date": "2024-01-15", "open": 72500, "high": 74000, "low": 72000, "close": 73000, "volume": 15000000},
{"date": "2024-01-14", "open": 71800, "high": 73200, "low": 71500, "close": 72500, "volume": 18500000},
{"date": "2024-01-13", "open": 70500, "high": 72000, "low": 70200, "close": 71800, "volume": 22000000},
{"date": "2024-01-12", "open": 69800, "high": 71000, "low": 69500, "close": 70500, "volume": 19800000},
{"date": "2024-01-11", "open": 68900, "high": 70200, "low": 68700, "close": 69800, "volume": 16200000},
{"date": "2024-01-10", "open": 67800, "high": 69200, "low": 67500, "close": 68900, "volume": 14800000},
{"date": "2024-01-09", "open": 66500, "high": 68100, "low": 66200, "close": 67800, "volume": 17300000},
{"date": "2024-01-08", "open": 65200, "high": 66800, "low": 65000, "close": 66500, "volume": 20100000}
]
@pytest.mark.asyncio
async def test_calculate_moving_averages(self, technical_analyzer, sample_ohlcv_data):
"""Test moving averages calculation (SMA, EMA)"""
ma_data = await technical_analyzer.calculate_moving_averages(
ohlcv_data=sample_ohlcv_data,
periods=[5, 10, 20]
)
assert ma_data is not None
assert isinstance(ma_data, dict)
# Should contain different MA types
assert "sma" in ma_data
assert "ema" in ma_data
# Should have calculations for different periods
for period in [5, 10, 20]:
assert period in ma_data["sma"]
assert period in ma_data["ema"]
# Values should be reasonable
assert ma_data["sma"][5] > 0
assert ma_data["ema"][5] > 0
assert ma_data["sma"][5] != ma_data["ema"][5] # SMA and EMA should differ
@pytest.mark.asyncio
async def test_calculate_rsi(self, technical_analyzer, sample_ohlcv_data):
"""Test RSI (Relative Strength Index) calculation"""
rsi_data = await technical_analyzer.calculate_rsi(
ohlcv_data=sample_ohlcv_data,
period=14
)
assert rsi_data is not None
assert isinstance(rsi_data, dict)
# Should contain RSI value and interpretation
assert "rsi" in rsi_data
assert "signal" in rsi_data
assert "condition" in rsi_data
# RSI should be between 0 and 100
assert 0 <= rsi_data["rsi"] <= 100
# Should have signal interpretation
assert rsi_data["signal"] in ["oversold", "neutral", "overbought"]
@pytest.mark.asyncio
async def test_calculate_macd(self, technical_analyzer, sample_ohlcv_data):
"""Test MACD calculation"""
macd_data = await technical_analyzer.calculate_macd(
ohlcv_data=sample_ohlcv_data,
fast_period=12,
slow_period=26,
signal_period=9
)
assert macd_data is not None
assert isinstance(macd_data, dict)
# Should contain MACD components
assert "macd_line" in macd_data
assert "signal_line" in macd_data
assert "histogram" in macd_data
assert "signal" in macd_data
# Should have buy/sell signal
assert macd_data["signal"] in ["bullish", "bearish", "neutral"]
@pytest.mark.asyncio
async def test_calculate_bollinger_bands(self, technical_analyzer, sample_ohlcv_data):
"""Test Bollinger Bands calculation"""
bb_data = await technical_analyzer.calculate_bollinger_bands(
ohlcv_data=sample_ohlcv_data,
period=20,
std_dev=2
)
assert bb_data is not None
assert isinstance(bb_data, dict)
# Should contain band components
assert "upper_band" in bb_data
assert "middle_band" in bb_data
assert "lower_band" in bb_data
assert "bandwidth" in bb_data
assert "position" in bb_data
# Bands should be properly ordered
assert bb_data["upper_band"] > bb_data["middle_band"]
assert bb_data["middle_band"] > bb_data["lower_band"]
# Position should indicate where current price is relative to bands
assert bb_data["position"] in ["above_upper", "above_middle", "below_middle", "below_lower"]
@pytest.mark.asyncio
async def test_calculate_stochastic(self, technical_analyzer, sample_ohlcv_data):
"""Test Stochastic Oscillator calculation"""
stoch_data = await technical_analyzer.calculate_stochastic(
ohlcv_data=sample_ohlcv_data,
k_period=14,
d_period=3
)
assert stoch_data is not None
assert isinstance(stoch_data, dict)
# Should contain stochastic components
assert "k_percent" in stoch_data
assert "d_percent" in stoch_data
assert "signal" in stoch_data
# Values should be between 0 and 100
assert 0 <= stoch_data["k_percent"] <= 100
assert 0 <= stoch_data["d_percent"] <= 100
# Should have signal interpretation
assert stoch_data["signal"] in ["oversold", "neutral", "overbought"]
@pytest.mark.asyncio
async def test_calculate_atr(self, technical_analyzer, sample_ohlcv_data):
"""Test Average True Range calculation"""
atr_data = await technical_analyzer.calculate_atr(
ohlcv_data=sample_ohlcv_data,
period=14
)
assert atr_data is not None
assert isinstance(atr_data, dict)
# Should contain ATR value and interpretation
assert "atr" in atr_data
assert "volatility_level" in atr_data
assert "trend_strength" in atr_data
# ATR should be positive
assert atr_data["atr"] > 0
# Should have volatility classification
assert atr_data["volatility_level"] in ["low", "medium", "high"]
@pytest.mark.asyncio
async def test_identify_support_resistance(self, technical_analyzer, sample_ohlcv_data):
"""Test support and resistance level identification"""
sr_data = await technical_analyzer.identify_support_resistance(
ohlcv_data=sample_ohlcv_data,
lookback_period=10
)
assert sr_data is not None
assert isinstance(sr_data, dict)
# Should contain support and resistance levels
assert "support_levels" in sr_data
assert "resistance_levels" in sr_data
assert "current_position" in sr_data
# Should have at least one level of each
assert len(sr_data["support_levels"]) > 0
assert len(sr_data["resistance_levels"]) > 0
# Support should be below current price, resistance above
current_price = sample_ohlcv_data[0]["close"]
for support in sr_data["support_levels"]:
assert support["level"] <= current_price
for resistance in sr_data["resistance_levels"]:
assert resistance["level"] >= current_price
@pytest.mark.asyncio
async def test_calculate_volume_indicators(self, technical_analyzer, sample_ohlcv_data):
"""Test volume indicators (OBV, Volume Profile)"""
volume_data = await technical_analyzer.calculate_volume_indicators(
ohlcv_data=sample_ohlcv_data
)
assert volume_data is not None
assert isinstance(volume_data, dict)
# Should contain volume indicators
assert "obv" in volume_data
assert "volume_trend" in volume_data
assert "accumulation_distribution" in volume_data
# Should have trend interpretation
assert volume_data["volume_trend"] in ["accumulation", "distribution", "neutral"]
@pytest.mark.asyncio
async def test_generate_trading_signals(self, technical_analyzer, sample_ohlcv_data):
"""Test trading signals generation"""
signals = await technical_analyzer.generate_trading_signals(
ohlcv_data=sample_ohlcv_data,
indicators=["rsi", "macd", "stochastic", "bollinger"]
)
assert signals is not None
assert isinstance(signals, dict)
# Should contain overall signal
assert "overall_signal" in signals
assert "signal_strength" in signals
assert "individual_signals" in signals
assert "entry_exit_points" in signals
# Overall signal should be one of the expected values
assert signals["overall_signal"] in ["strong_buy", "buy", "hold", "sell", "strong_sell"]
# Signal strength should be 0-100
assert 0 <= signals["signal_strength"] <= 100
# Should have individual indicator signals
for indicator in ["rsi", "macd", "stochastic", "bollinger"]:
assert indicator in signals["individual_signals"]
@pytest.mark.asyncio
async def test_comprehensive_technical_analysis(self, technical_analyzer, sample_ohlcv_data):
"""Test comprehensive technical analysis"""
analysis = await technical_analyzer.comprehensive_analysis(
ohlcv_data=sample_ohlcv_data,
include_all_indicators=True
)
assert analysis is not None
assert isinstance(analysis, dict)
# Should contain all major analysis sections
assert "moving_averages" in analysis
assert "momentum_indicators" in analysis
assert "volatility_indicators" in analysis
assert "volume_analysis" in analysis
assert "support_resistance" in analysis
assert "trading_signals" in analysis
assert "technical_rating" in analysis
# Should have overall technical rating
assert "score" in analysis["technical_rating"]
assert "recommendation" in analysis["technical_rating"]
assert 0 <= analysis["technical_rating"]["score"] <= 100
@pytest.mark.asyncio
async def test_pattern_recognition(self, technical_analyzer, sample_ohlcv_data):
"""Test chart pattern recognition"""
patterns = await technical_analyzer.recognize_patterns(
ohlcv_data=sample_ohlcv_data,
pattern_types=["trend_lines", "triangles", "head_shoulders", "double_tops"]
)
assert patterns is not None
assert isinstance(patterns, list)
# Each pattern should have required fields
for pattern in patterns:
assert "type" in pattern
assert "confidence" in pattern
assert "prediction" in pattern
assert "start_date" in pattern
assert "end_date" in pattern
@pytest.mark.asyncio
async def test_error_handling_insufficient_data(self, technical_analyzer):
"""Test error handling for insufficient data"""
# Test with too little data
insufficient_data = [
{"date": "2024-01-15", "open": 72500, "high": 74000, "low": 72000, "close": 73000, "volume": 15000000}
]
with pytest.raises(InsufficientDataError):
await technical_analyzer.calculate_moving_averages(insufficient_data, periods=[20])
@pytest.mark.asyncio
async def test_error_handling_invalid_parameters(self, technical_analyzer, sample_ohlcv_data):
"""Test error handling for invalid parameters"""
# Test with invalid period
with pytest.raises(MCPStockDetailsError):
await technical_analyzer.calculate_rsi(sample_ohlcv_data, period=0)
# Test with invalid standard deviation
with pytest.raises(MCPStockDetailsError):
await technical_analyzer.calculate_bollinger_bands(sample_ohlcv_data, period=20, std_dev=-1)