"""
Unit tests for Technical Indicators functionality
TDD Red Phase: Write failing tests for technical analysis
"""
import pytest
import asyncio
from unittest.mock import Mock, AsyncMock, patch
from datetime import datetime, date, timedelta
from typing import Any, Dict, List, Optional
# Test imports - initially will fail (TDD Red phase)
from src.server import MCPStockDetailsServer
from src.tools.technical_tools import TechnicalAnalyzer
from src.exceptions import MCPStockDetailsError, InsufficientDataError
class TestTechnicalIndicators:
"""Test cases for technical indicators functionality"""
@pytest.fixture
def server_with_technical(self):
"""Create server instance with technical analyzer"""
server = MCPStockDetailsServer()
# This will fail initially - Red phase
server.technical_analyzer = TechnicalAnalyzer()
return server
@pytest.fixture
def sample_price_data(self):
"""Sample price data for technical analysis"""
return {
"005930": { # Samsung Electronics
"current_price": 73000,
"daily_data": [
{"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}
],
"weekly_data": [
{"week": "2024-W03", "open": 68900, "high": 74000, "low": 68700, "close": 73000, "volume": 91500000},
{"week": "2024-W02", "open": 67200, "high": 69500, "low": 66800, "close": 68900, "volume": 87200000}
],
"monthly_data": [
{"month": "2024-01", "open": 67200, "high": 74000, "low": 66800, "close": 73000, "volume": 178700000},
{"month": "2023-12", "open": 65800, "high": 68500, "low": 64200, "close": 67200, "volume": 195300000}
]
}
}
@pytest.mark.asyncio
async def test_get_technical_indicators_tool_registration(self, server_with_technical):
"""Test that get_technical_indicators tool is properly registered"""
tools = await server_with_technical.list_tools()
tool_names = [tool.name for tool in tools]
assert "get_technical_indicators" in tool_names
# Check tool description and parameters
tech_tool = next(tool for tool in tools if tool.name == "get_technical_indicators")
assert "technical" in tech_tool.description.lower()
assert "indicators" in tech_tool.description.lower()
assert "company_code" in tech_tool.inputSchema["properties"]
@pytest.mark.asyncio
async def test_basic_moving_averages(self, server_with_technical, sample_price_data):
"""Test basic moving averages calculation"""
company_code = "005930"
with patch.object(server_with_technical.technical_analyzer, 'get_price_data') as mock_price:
mock_price.return_value = sample_price_data[company_code]
result = await server_with_technical._handle_get_technical_indicators({
"company_code": company_code,
"include_moving_averages": True
})
assert result is not None
assert len(result) > 0
content = result[0].text
# Should include moving averages
assert "MOVING AVERAGES" in content
assert "SMA" in content or "Simple Moving Average" in content
assert "EMA" in content or "Exponential Moving Average" in content
assert "5-day" in content or "20-day" in content
@pytest.mark.asyncio
async def test_momentum_indicators(self, server_with_technical, sample_price_data):
"""Test momentum indicators (RSI, MACD, Stochastic)"""
company_code = "005930"
with patch.object(server_with_technical.technical_analyzer, 'get_price_data') as mock_price:
mock_price.return_value = sample_price_data[company_code]
result = await server_with_technical._handle_get_technical_indicators({
"company_code": company_code,
"include_momentum": True
})
assert result is not None
content = result[0].text
# Should include momentum indicators
assert "MOMENTUM INDICATORS" in content
assert "RSI" in content
assert "MACD" in content
assert "Stochastic" in content or "Stoch" in content
@pytest.mark.asyncio
async def test_volatility_indicators(self, server_with_technical):
"""Test volatility indicators (Bollinger Bands, ATR)"""
company_code = "005930"
result = await server_with_technical._handle_get_technical_indicators({
"company_code": company_code,
"include_volatility": True
})
assert result is not None
content = result[0].text
# Should include volatility indicators
assert "VOLATILITY INDICATORS" in content
assert "Bollinger Bands" in content
assert "ATR" in content or "Average True Range" in content
assert "Upper Band" in content
assert "Lower Band" in content
@pytest.mark.asyncio
async def test_trend_indicators(self, server_with_technical):
"""Test trend indicators (ADX, Parabolic SAR)"""
company_code = "005930"
result = await server_with_technical._handle_get_technical_indicators({
"company_code": company_code,
"include_trend": True
})
assert result is not None
content = result[0].text
# Should include trend indicators
assert "TREND INDICATORS" in content
assert "ADX" in content or "Directional Movement" in content
assert "Parabolic SAR" in content or "PSAR" in content
assert "Trend Strength" in content
@pytest.mark.asyncio
async def test_volume_indicators(self, server_with_technical):
"""Test volume indicators (OBV, Volume Profile)"""
company_code = "005930"
result = await server_with_technical._handle_get_technical_indicators({
"company_code": company_code,
"include_volume": True
})
assert result is not None
content = result[0].text
# Should include volume indicators
assert "VOLUME INDICATORS" in content
assert "OBV" in content or "On-Balance Volume" in content
assert "Volume Profile" in content
assert "Accumulation" in content or "Distribution" in content
@pytest.mark.asyncio
async def test_support_resistance_levels(self, server_with_technical, sample_price_data):
"""Test support and resistance level identification"""
company_code = "005930"
with patch.object(server_with_technical.technical_analyzer, 'get_price_data') as mock_price:
mock_price.return_value = sample_price_data[company_code]
result = await server_with_technical._handle_get_technical_indicators({
"company_code": company_code,
"include_support_resistance": True
})
assert result is not None
content = result[0].text
# Should include support/resistance levels
assert "SUPPORT & RESISTANCE" in content
assert "Support Level" in content
assert "Resistance Level" in content
assert "Key Levels" in content
@pytest.mark.asyncio
async def test_chart_patterns(self, server_with_technical):
"""Test chart pattern recognition"""
company_code = "005930"
result = await server_with_technical._handle_get_technical_indicators({
"company_code": company_code,
"include_patterns": True
})
assert result is not None
content = result[0].text
# Should include chart patterns
assert "CHART PATTERNS" in content
assert "Pattern Recognition" in content
assert "Bullish" in content or "Bearish" in content
@pytest.mark.asyncio
async def test_technical_signals(self, server_with_technical, sample_price_data):
"""Test trading signals generation"""
company_code = "005930"
with patch.object(server_with_technical.technical_analyzer, 'get_price_data') as mock_price:
mock_price.return_value = sample_price_data[company_code]
result = await server_with_technical._handle_get_technical_indicators({
"company_code": company_code,
"include_signals": True
})
assert result is not None
content = result[0].text
# Should include trading signals
assert "TRADING SIGNALS" in content
assert "BUY" in content or "SELL" in content or "HOLD" in content
assert "Signal Strength" in content
assert "Entry Point" in content or "Exit Point" in content
@pytest.mark.asyncio
async def test_comprehensive_technical_analysis(self, server_with_technical, sample_price_data):
"""Test comprehensive technical analysis report"""
company_code = "005930"
with patch.object(server_with_technical.technical_analyzer, 'get_price_data') as mock_price:
mock_price.return_value = sample_price_data[company_code]
result = await server_with_technical._handle_get_technical_indicators({
"company_code": company_code,
"analysis_type": "comprehensive"
})
assert result is not None
content = result[0].text
# Should include comprehensive analysis
assert "COMPREHENSIVE TECHNICAL ANALYSIS" in content
assert "Overall Technical Rating" in content
assert "Market Sentiment" in content
assert "Key Insights" in content
@pytest.mark.asyncio
async def test_technical_indicators_error_handling(self, server_with_technical):
"""Test error handling for technical indicators"""
# Test invalid company code
with pytest.raises(MCPStockDetailsError):
await server_with_technical._handle_get_technical_indicators({
"company_code": "",
"include_moving_averages": True
})
# Test insufficient price data
with pytest.raises(InsufficientDataError):
await server_with_technical._handle_get_technical_indicators({
"company_code": "999999", # Non-existent company
"include_moving_averages": True
})
