325 lines
12 KiB
Python
325 lines
12 KiB
Python
|
"""
|
||
|
|
Safety net tests for technical indicators module.
|
||
|
|
|
||
|
|
These tests ensure that the core functionality of the indicators module
|
||
|
|
remains intact during refactoring.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import pytest
|
||
|
|
from datetime import datetime, timezone, timedelta
|
||
|
|
from decimal import Decimal
|
||
|
|
import pandas as pd
|
||
|
|
import numpy as np
|
||
|
|
|
||
|
|
from data.common.indicators import (
|
||
|
|
TechnicalIndicators,
|
||
|
|
IndicatorResult,
|
||
|
|
create_default_indicators_config,
|
||
|
|
validate_indicator_config
|
||
|
|
)
|
||
|
|
from data.common.data_types import OHLCVCandle
|
||
|
|
|
||
|
|
|
||
|
|
class TestTechnicalIndicatorsSafety:
|
||
|
|
"""Safety net test suite for TechnicalIndicators class."""
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def sample_candles(self):
|
||
|
|
"""Create sample OHLCV candles for testing."""
|
||
|
|
candles = []
|
||
|
|
base_time = datetime(2024, 1, 1, 9, 0, 0, tzinfo=timezone.utc)
|
||
|
|
|
||
|
|
# Create 30 candles with realistic price movement
|
||
|
|
prices = [100.0, 101.0, 102.5, 101.8, 103.0, 104.2, 103.8, 105.0, 104.5, 106.0,
|
||
|
|
107.5, 108.0, 107.2, 109.0, 108.5, 110.0, 109.8, 111.0, 110.5, 112.0,
|
||
|
|
111.8, 113.0, 112.5, 114.0, 113.2, 115.0, 114.8, 116.0, 115.5, 117.0]
|
||
|
|
|
||
|
|
for i, price in enumerate(prices):
|
||
|
|
candle = OHLCVCandle(
|
||
|
|
symbol='BTC-USDT',
|
||
|
|
timeframe='1m',
|
||
|
|
start_time=base_time + timedelta(minutes=i),
|
||
|
|
end_time=base_time + timedelta(minutes=i+1),
|
||
|
|
open=Decimal(str(price - 0.2)),
|
||
|
|
high=Decimal(str(price + 0.5)),
|
||
|
|
low=Decimal(str(price - 0.5)),
|
||
|
|
close=Decimal(str(price)),
|
||
|
|
volume=Decimal('1000'),
|
||
|
|
trade_count=10,
|
||
|
|
exchange='test',
|
||
|
|
is_complete=True
|
||
|
|
)
|
||
|
|
candles.append(candle)
|
||
|
|
|
||
|
|
return candles
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def sparse_candles(self):
|
||
|
|
"""Create sample OHLCV candles with time gaps for testing."""
|
||
|
|
candles = []
|
||
|
|
base_time = datetime(2024, 1, 1, 9, 0, 0, tzinfo=timezone.utc)
|
||
|
|
|
||
|
|
# Create 15 candles with gaps (every other minute)
|
||
|
|
prices = [100.0, 102.5, 104.2, 105.0, 106.0,
|
||
|
|
108.0, 109.0, 110.0, 111.0, 112.0,
|
||
|
|
113.0, 114.0, 115.0, 116.0, 117.0]
|
||
|
|
|
||
|
|
for i, price in enumerate(prices):
|
||
|
|
# Create 2-minute gaps between candles
|
||
|
|
candle = OHLCVCandle(
|
||
|
|
symbol='BTC-USDT',
|
||
|
|
timeframe='1m',
|
||
|
|
start_time=base_time + timedelta(minutes=i*2),
|
||
|
|
end_time=base_time + timedelta(minutes=(i*2)+1),
|
||
|
|
open=Decimal(str(price - 0.2)),
|
||
|
|
high=Decimal(str(price + 0.5)),
|
||
|
|
low=Decimal(str(price - 0.5)),
|
||
|
|
close=Decimal(str(price)),
|
||
|
|
volume=Decimal('1000'),
|
||
|
|
trade_count=10,
|
||
|
|
exchange='test',
|
||
|
|
is_complete=True
|
||
|
|
)
|
||
|
|
candles.append(candle)
|
||
|
|
|
||
|
|
return candles
|
||
|
|
|
||
|
|
@pytest.fixture
|
||
|
|
def indicators(self):
|
||
|
|
"""Create TechnicalIndicators instance."""
|
||
|
|
return TechnicalIndicators()
|
||
|
|
|
||
|
|
def test_initialization(self, indicators):
|
||
|
|
"""Test indicator calculator initialization."""
|
||
|
|
assert isinstance(indicators, TechnicalIndicators)
|
||
|
|
|
||
|
|
def test_prepare_dataframe_from_list(self, indicators, sample_candles):
|
||
|
|
"""Test DataFrame preparation from OHLCV candles."""
|
||
|
|
df = indicators._prepare_dataframe_from_list(sample_candles)
|
||
|
|
assert isinstance(df, pd.DataFrame)
|
||
|
|
assert not df.empty
|
||
|
|
assert len(df) == len(sample_candles)
|
||
|
|
assert 'close' in df.columns
|
||
|
|
assert 'timestamp' in df.index.names
|
||
|
|
|
||
|
|
def test_prepare_dataframe_empty(self, indicators):
|
||
|
|
"""Test DataFrame preparation with empty candles list."""
|
||
|
|
df = indicators._prepare_dataframe_from_list([])
|
||
|
|
assert isinstance(df, pd.DataFrame)
|
||
|
|
assert df.empty
|
||
|
|
|
||
|
|
def test_sma_calculation(self, indicators, sample_candles):
|
||
|
|
"""Test Simple Moving Average calculation."""
|
||
|
|
period = 5
|
||
|
|
df = indicators._prepare_dataframe_from_list(sample_candles)
|
||
|
|
results = indicators.sma(df, period)
|
||
|
|
|
||
|
|
assert len(results) > 0
|
||
|
|
assert isinstance(results[0], IndicatorResult)
|
||
|
|
assert 'sma' in results[0].values
|
||
|
|
assert results[0].metadata['period'] == period
|
||
|
|
|
||
|
|
def test_sma_insufficient_data(self, indicators, sample_candles):
|
||
|
|
"""Test SMA with insufficient data."""
|
||
|
|
period = 50 # More than available candles
|
||
|
|
df = indicators._prepare_dataframe_from_list(sample_candles)
|
||
|
|
results = indicators.sma(df, period)
|
||
|
|
assert len(results) == 0
|
||
|
|
|
||
|
|
def test_ema_calculation(self, indicators, sample_candles):
|
||
|
|
"""Test Exponential Moving Average calculation."""
|
||
|
|
period = 10
|
||
|
|
df = indicators._prepare_dataframe_from_list(sample_candles)
|
||
|
|
results = indicators.ema(df, period)
|
||
|
|
|
||
|
|
assert len(results) > 0
|
||
|
|
assert isinstance(results[0], IndicatorResult)
|
||
|
|
assert 'ema' in results[0].values
|
||
|
|
assert results[0].metadata['period'] == period
|
||
|
|
|
||
|
|
def test_rsi_calculation(self, indicators, sample_candles):
|
||
|
|
"""Test Relative Strength Index calculation."""
|
||
|
|
period = 14
|
||
|
|
df = indicators._prepare_dataframe_from_list(sample_candles)
|
||
|
|
results = indicators.rsi(df, period)
|
||
|
|
|
||
|
|
assert len(results) > 0
|
||
|
|
assert isinstance(results[0], IndicatorResult)
|
||
|
|
assert 'rsi' in results[0].values
|
||
|
|
assert results[0].metadata['period'] == period
|
||
|
|
assert 0 <= results[0].values['rsi'] <= 100
|
||
|
|
|
||
|
|
def test_macd_calculation(self, indicators, sample_candles):
|
||
|
|
"""Test MACD calculation."""
|
||
|
|
fast_period = 12
|
||
|
|
slow_period = 26
|
||
|
|
signal_period = 9
|
||
|
|
df = indicators._prepare_dataframe_from_list(sample_candles)
|
||
|
|
results = indicators.macd(df, fast_period, slow_period, signal_period)
|
||
|
|
|
||
|
|
# MACD should start producing results after slow_period periods
|
||
|
|
assert len(results) > 0
|
||
|
|
|
||
|
|
if results: # Only test if we have results
|
||
|
|
first_result = results[0]
|
||
|
|
assert isinstance(first_result, IndicatorResult)
|
||
|
|
assert 'macd' in first_result.values
|
||
|
|
assert 'signal' in first_result.values
|
||
|
|
assert 'histogram' in first_result.values
|
||
|
|
|
||
|
|
# Histogram should equal MACD - Signal
|
||
|
|
expected_histogram = first_result.values['macd'] - first_result.values['signal']
|
||
|
|
assert abs(first_result.values['histogram'] - expected_histogram) < 0.001
|
||
|
|
|
||
|
|
def test_bollinger_bands_calculation(self, indicators, sample_candles):
|
||
|
|
"""Test Bollinger Bands calculation."""
|
||
|
|
period = 20
|
||
|
|
std_dev = 2.0
|
||
|
|
df = indicators._prepare_dataframe_from_list(sample_candles)
|
||
|
|
results = indicators.bollinger_bands(df, period, std_dev)
|
||
|
|
|
||
|
|
assert len(results) > 0
|
||
|
|
assert isinstance(results[0], IndicatorResult)
|
||
|
|
assert 'upper_band' in results[0].values
|
||
|
|
assert 'middle_band' in results[0].values
|
||
|
|
assert 'lower_band' in results[0].values
|
||
|
|
assert results[0].metadata['period'] == period
|
||
|
|
assert results[0].metadata['std_dev'] == std_dev
|
||
|
|
|
||
|
|
def test_sparse_data_handling(self, indicators, sparse_candles):
|
||
|
|
"""Test indicators with sparse data (time gaps)."""
|
||
|
|
period = 5
|
||
|
|
df = indicators._prepare_dataframe_from_list(sparse_candles)
|
||
|
|
sma_results = indicators.sma(df, period)
|
||
|
|
|
||
|
|
assert len(sma_results) > 0
|
||
|
|
# Verify that gaps are preserved (no interpolation)
|
||
|
|
timestamps = [r.timestamp for r in sma_results]
|
||
|
|
for i in range(1, len(timestamps)):
|
||
|
|
time_diff = timestamps[i] - timestamps[i-1]
|
||
|
|
assert time_diff >= timedelta(minutes=1)
|
||
|
|
|
||
|
|
def test_calculate_multiple_indicators(self, indicators, sample_candles):
|
||
|
|
"""Test calculating multiple indicators at once."""
|
||
|
|
config = {
|
||
|
|
'sma_10': {'type': 'sma', 'period': 10},
|
||
|
|
'ema_12': {'type': 'ema', 'period': 12},
|
||
|
|
'rsi_14': {'type': 'rsi', 'period': 14},
|
||
|
|
'macd': {'type': 'macd'},
|
||
|
|
'bb_20': {'type': 'bollinger_bands', 'period': 20}
|
||
|
|
}
|
||
|
|
|
||
|
|
df = indicators._prepare_dataframe_from_list(sample_candles)
|
||
|
|
results = indicators.calculate_multiple_indicators(df, config)
|
||
|
|
|
||
|
|
assert len(results) == len(config)
|
||
|
|
assert 'sma_10' in results
|
||
|
|
assert 'ema_12' in results
|
||
|
|
assert 'rsi_14' in results
|
||
|
|
assert 'macd' in results
|
||
|
|
assert 'bb_20' in results
|
||
|
|
|
||
|
|
# Check that each indicator has appropriate results
|
||
|
|
assert len(results['sma_10']) > 0
|
||
|
|
assert len(results['ema_12']) > 0
|
||
|
|
assert len(results['rsi_14']) > 0
|
||
|
|
assert len(results['macd']) > 0
|
||
|
|
assert len(results['bb_20']) > 0
|
||
|
|
|
||
|
|
def test_different_price_columns(self, indicators, sample_candles):
|
||
|
|
"""Test indicators with different price columns."""
|
||
|
|
df = indicators._prepare_dataframe_from_list(sample_candles)
|
||
|
|
|
||
|
|
# Test SMA with 'high' price column
|
||
|
|
sma_high = indicators.sma(df, 5, price_column='high')
|
||
|
|
assert len(sma_high) > 0
|
||
|
|
|
||
|
|
# Test SMA with 'low' price column
|
||
|
|
sma_low = indicators.sma(df, 5, price_column='low')
|
||
|
|
assert len(sma_low) > 0
|
||
|
|
|
||
|
|
# Values should be different
|
||
|
|
assert sma_high[0].values['sma'] != sma_low[0].values['sma']
|
||
|
|
|
||
|
|
|
||
|
|
class TestIndicatorHelperFunctions:
|
||
|
|
"""Test suite for indicator helper functions."""
|
||
|
|
|
||
|
|
def test_create_default_indicators_config(self):
|
||
|
|
"""Test default indicator configuration creation."""
|
||
|
|
config = create_default_indicators_config()
|
||
|
|
assert isinstance(config, dict)
|
||
|
|
assert len(config) > 0
|
||
|
|
assert 'sma_20' in config
|
||
|
|
assert 'ema_12' in config
|
||
|
|
assert 'rsi_14' in config
|
||
|
|
assert 'macd_default' in config
|
||
|
|
assert 'bollinger_bands_20' in config
|
||
|
|
|
||
|
|
def test_validate_indicator_config_valid(self):
|
||
|
|
"""Test indicator configuration validation with valid config."""
|
||
|
|
valid_configs = [
|
||
|
|
{'type': 'sma', 'period': 20},
|
||
|
|
{'type': 'ema', 'period': 12},
|
||
|
|
{'type': 'rsi', 'period': 14},
|
||
|
|
{'type': 'macd'},
|
||
|
|
{'type': 'bollinger_bands', 'period': 20, 'std_dev': 2.0}
|
||
|
|
]
|
||
|
|
|
||
|
|
for config in valid_configs:
|
||
|
|
assert validate_indicator_config(config)
|
||
|
|
|
||
|
|
def test_validate_indicator_config_invalid(self):
|
||
|
|
"""Test indicator configuration validation with invalid config."""
|
||
|
|
invalid_configs = [
|
||
|
|
{}, # Empty config
|
||
|
|
{'type': 'unknown'}, # Invalid type
|
||
|
|
{'type': 'sma', 'period': -1}, # Invalid period
|
||
|
|
{'type': 'bollinger_bands', 'std_dev': -1}, # Invalid std_dev
|
||
|
|
{'type': 'sma', 'period': 'not_a_number'} # Wrong type for period
|
||
|
|
]
|
||
|
|
|
||
|
|
for config in invalid_configs:
|
||
|
|
assert not validate_indicator_config(config)
|
||
|
|
|
||
|
|
|
||
|
|
class TestIndicatorResultDataClass:
|
||
|
|
"""Test suite for IndicatorResult dataclass."""
|
||
|
|
|
||
|
|
def test_indicator_result_creation(self):
|
||
|
|
"""Test IndicatorResult creation with all fields."""
|
||
|
|
timestamp = datetime.now(timezone.utc)
|
||
|
|
values = {'sma': 100.0}
|
||
|
|
metadata = {'period': 20}
|
||
|
|
|
||
|
|
result = IndicatorResult(
|
||
|
|
timestamp=timestamp,
|
||
|
|
symbol='BTC-USDT',
|
||
|
|
timeframe='1m',
|
||
|
|
values=values,
|
||
|
|
metadata=metadata
|
||
|
|
)
|
||
|
|
|
||
|
|
assert result.timestamp == timestamp
|
||
|
|
assert result.symbol == 'BTC-USDT'
|
||
|
|
assert result.timeframe == '1m'
|
||
|
|
assert result.values == values
|
||
|
|
assert result.metadata == metadata
|
||
|
|
|
||
|
|
def test_indicator_result_without_metadata(self):
|
||
|
|
"""Test IndicatorResult creation without optional metadata."""
|
||
|
|
timestamp = datetime.now(timezone.utc)
|
||
|
|
values = {'sma': 100.0}
|
||
|
|
|
||
|
|
result = IndicatorResult(
|
||
|
|
timestamp=timestamp,
|
||
|
|
symbol='BTC-USDT',
|
||
|
|
timeframe='1m',
|
||
|
|
values=values
|
||
|
|
)
|
||
|
|
|
||
|
|
assert result.timestamp == timestamp
|
||
|
|
assert result.symbol == 'BTC-USDT'
|
||
|
|
assert result.timeframe == '1m'
|
||
|
|
assert result.values == values
|
||
|
|
assert result.metadata is None
|