"""멀티 타임프레임 분석기 테스트"""
import pytest
from datetime import datetime, timedelta
from src.analytics.multi_timeframe_analyzer import MultiTimeframeAnalyzer
from src.exceptions import TimeframeError, AnalysisError, InsufficientDataError
class TestMultiTimeframeAnalyzer:
"""멀티 타임프레임 분석기 테스트 클래스"""
@pytest.fixture
def analyzer_config(self):
"""분석기 설정"""
return {
"timeframes": ["1m", "5m", "15m", "1h", "4h", "1d"],
"aggregation_methods": {
"volume": "sum",
"close": "last",
"high": "max",
"low": "min",
"open": "first"
},
"analysis_types": [
"trend_alignment",
"momentum_divergence",
"volume_profile",
"support_resistance"
],
"correlation_thresholds": {
"strong": 0.8,
"moderate": 0.6,
"weak": 0.3
}
}
@pytest.fixture
def multi_timeframe_analyzer(self, analyzer_config):
"""멀티 타임프레임 분석기 인스턴스"""
return MultiTimeframeAnalyzer(analyzer_config)
@pytest.fixture
def sample_ohlcv_data(self):
"""샘플 OHLCV 데이터 (1분봉)"""
data = []
base_price = 50000
base_time = datetime(2024, 1, 1, 0, 0, 0)
for i in range(1440): # 24시간 = 1440분
time = base_time + timedelta(minutes=i)
# 가격 변동 시뮬레이션
trend = (i // 60) * 100 # 시간별 트렌드
wave = 500 * ((i % 60) / 60) # 분별 변동
noise = (i % 5) * 20 # 노이즈
close = base_price + trend + wave + noise
high = close + abs(noise) * 2
low = close - abs(noise) * 1.5
open_price = close - noise
volume = 1000000 + (i % 30) * 50000
data.append({
"timestamp": time.isoformat(),
"open": open_price,
"high": high,
"low": low,
"close": close,
"volume": volume
})
return data
@pytest.fixture
def multi_symbol_data(self):
"""여러 심볼의 데이터"""
symbols = ["BTC/USD", "ETH/USD", "SOL/USD"]
data = {}
for symbol in symbols:
base_price = 50000 if "BTC" in symbol else 3000 if "ETH" in symbol else 100
symbol_data = []
for i in range(100):
symbol_data.append({
"timestamp": (datetime(2024, 1, 1, 0, 0, 0) + timedelta(hours=i)).isoformat(),
"symbol": symbol,
"close": base_price + i * 10,
"volume": 1000000 + i * 10000
})
data[symbol] = symbol_data
return data
@pytest.mark.asyncio
async def test_timeframe_aggregation(self, multi_timeframe_analyzer, sample_ohlcv_data):
"""타임프레임 집계 테스트"""
# 5분봉으로 집계
aggregated_5m = await multi_timeframe_analyzer.aggregate_to_timeframe(
sample_ohlcv_data,
target_timeframe="5m"
)
# 결과 검증
assert len(aggregated_5m) == 288 # 24시간 = 288개 5분봉
# 첫 번째 5분봉 검증
first_candle = aggregated_5m[0]
assert "open" in first_candle
assert "high" in first_candle
assert "low" in first_candle
assert "close" in first_candle
assert "volume" in first_candle
# 집계 규칙 검증
# open은 첫 번째 값
assert first_candle["open"] == sample_ohlcv_data[0]["open"]
# close는 마지막 값
assert first_candle["close"] == sample_ohlcv_data[4]["close"]
# high는 최대값
max_high = max(d["high"] for d in sample_ohlcv_data[:5])
assert first_candle["high"] == max_high
# low는 최소값
min_low = min(d["low"] for d in sample_ohlcv_data[:5])
assert first_candle["low"] == min_low
# volume은 합계
sum_volume = sum(d["volume"] for d in sample_ohlcv_data[:5])
assert first_candle["volume"] == sum_volume
@pytest.mark.asyncio
async def test_multi_timeframe_analysis(self, multi_timeframe_analyzer, sample_ohlcv_data):
"""멀티 타임프레임 분석 테스트"""
# 여러 타임프레임 분석
analysis_results = await multi_timeframe_analyzer.analyze_timeframes(
sample_ohlcv_data,
timeframes=["5m", "15m", "1h"]
)
# 결과 검증
assert "timeframe_data" in analysis_results
assert "trend_alignment" in analysis_results
assert "momentum_analysis" in analysis_results
assert "volume_profile" in analysis_results
# 타임프레임별 데이터 확인
tf_data = analysis_results["timeframe_data"]
assert "5m" in tf_data
assert "15m" in tf_data
assert "1h" in tf_data
# 트렌드 정렬 확인
trend_alignment = analysis_results["trend_alignment"]
assert "alignment_score" in trend_alignment
assert "timeframe_trends" in trend_alignment
assert 0 <= trend_alignment["alignment_score"] <= 1
@pytest.mark.asyncio
async def test_trend_alignment_detection(self, multi_timeframe_analyzer, sample_ohlcv_data):
"""트렌드 정렬 감지 테스트"""
# 트렌드 정렬 분석
trend_alignment = await multi_timeframe_analyzer.detect_trend_alignment(
sample_ohlcv_data,
timeframes=["5m", "15m", "1h"],
lookback_periods={"5m": 20, "15m": 20, "1h": 20}
)
# 결과 검증
assert "aligned" in trend_alignment
assert "direction" in trend_alignment
assert "strength" in trend_alignment
assert "timeframe_trends" in trend_alignment
# 타임프레임별 트렌드
tf_trends = trend_alignment["timeframe_trends"]
for tf in ["5m", "15m", "1h"]:
assert tf in tf_trends
assert "direction" in tf_trends[tf]
assert "strength" in tf_trends[tf]
assert tf_trends[tf]["direction"] in ["up", "down", "neutral"]
@pytest.mark.asyncio
async def test_momentum_divergence_analysis(self, multi_timeframe_analyzer, sample_ohlcv_data):
"""모멘텀 다이버전스 분석 테스트"""
# 모멘텀 다이버전스 분석
divergence_results = await multi_timeframe_analyzer.analyze_momentum_divergence(
sample_ohlcv_data,
timeframes=["5m", "15m"],
momentum_indicator="rsi"
)
# 결과 검증
assert "divergences" in divergence_results
assert "momentum_values" in divergence_results
assert "divergence_strength" in divergence_results
# 다이버전스 검증
divergences = divergence_results["divergences"]
for div in divergences:
assert "type" in div # bullish_divergence, bearish_divergence
assert "timeframe" in div
assert "index" in div
assert "strength" in div
@pytest.mark.asyncio
async def test_volume_profile_analysis(self, multi_timeframe_analyzer, sample_ohlcv_data):
"""볼륨 프로파일 분석 테스트"""
# 볼륨 프로파일 분석
volume_profile = await multi_timeframe_analyzer.analyze_volume_profile(
sample_ohlcv_data,
timeframes=["5m", "15m", "1h"],
profile_bins=20
)
# 결과 검증
assert "price_levels" in volume_profile
assert "volume_distribution" in volume_profile
assert "poc" in volume_profile # Point of Control
assert "value_area" in volume_profile
# POC 검증
poc = volume_profile["poc"]
assert "price" in poc
assert "volume" in poc
# 가치 영역 검증
value_area = volume_profile["value_area"]
assert "high" in value_area
assert "low" in value_area
assert "volume_percentage" in value_area
assert 0 <= value_area["volume_percentage"] <= 1
@pytest.mark.asyncio
async def test_support_resistance_across_timeframes(self, multi_timeframe_analyzer, sample_ohlcv_data):
"""타임프레임별 지지/저항 분석 테스트"""
# 지지/저항 레벨 분석
sr_levels = await multi_timeframe_analyzer.find_support_resistance_levels(
sample_ohlcv_data,
timeframes=["5m", "15m", "1h"],
min_touches=2
)
# 결과 검증
assert "levels" in sr_levels
assert "timeframe_levels" in sr_levels
assert "confluence_zones" in sr_levels
# 레벨 검증
levels = sr_levels["levels"]
assert len(levels) > 0
for level in levels:
assert "price" in level
assert "type" in level # support or resistance
assert "strength" in level
assert "timeframes" in level
assert len(level["timeframes"]) > 0
@pytest.mark.asyncio
async def test_correlation_matrix_generation(self, multi_timeframe_analyzer, multi_symbol_data):
"""상관관계 매트릭스 생성 테스트"""
# 상관관계 매트릭스 계산
correlation_matrix = await multi_timeframe_analyzer.calculate_correlation_matrix(
multi_symbol_data,
timeframe="1h",
correlation_window=24
)
# 결과 검증
assert "matrix" in correlation_matrix
assert "symbols" in correlation_matrix
assert "significant_correlations" in correlation_matrix
# 매트릭스 검증
matrix = correlation_matrix["matrix"]
symbols = correlation_matrix["symbols"]
assert len(matrix) == len(symbols)
assert all(len(row) == len(symbols) for row in matrix)
# 대각선은 1이어야 함
for i in range(len(symbols)):
assert matrix[i][i] == 1.0
# 상관관계는 -1과 1 사이
for row in matrix:
for corr in row:
assert -1 <= corr <= 1
@pytest.mark.asyncio
async def test_timeframe_strength_ranking(self, multi_timeframe_analyzer, sample_ohlcv_data):
"""타임프레임 강도 순위 테스트"""
# 타임프레임별 강도 평가
strength_ranking = await multi_timeframe_analyzer.rank_timeframe_strength(
sample_ohlcv_data,
timeframes=["5m", "15m", "1h"],
criteria=["trend", "momentum", "volume"]
)
# 결과 검증
assert "rankings" in strength_ranking
assert "scores" in strength_ranking
assert "strongest_timeframe" in strength_ranking
# 순위 검증
rankings = strength_ranking["rankings"]
assert len(rankings) == 3
for rank in rankings:
assert "timeframe" in rank
assert "score" in rank
assert "breakdown" in rank
assert 0 <= rank["score"] <= 1
@pytest.mark.asyncio
async def test_market_regime_identification(self, multi_timeframe_analyzer, sample_ohlcv_data):
"""시장 체제 식별 테스트"""
# 시장 체제 분석
market_regime = await multi_timeframe_analyzer.identify_market_regime(
sample_ohlcv_data,
timeframes=["15m", "1h", "4h"]
)
# 결과 검증
assert "regime" in market_regime
assert "confidence" in market_regime
assert "characteristics" in market_regime
assert "timeframe_regimes" in market_regime
# 체제 타입 검증
assert market_regime["regime"] in ["trending", "ranging", "volatile", "quiet"]
assert 0 <= market_regime["confidence"] <= 1
# 특성 검증
characteristics = market_regime["characteristics"]
assert "volatility" in characteristics
assert "trend_strength" in characteristics
assert "volume_profile" in characteristics
@pytest.mark.asyncio
async def test_divergence_confluence_detection(self, multi_timeframe_analyzer, sample_ohlcv_data):
"""다이버전스 컨플루언스 감지 테스트"""
# 다이버전스 컨플루언스 분석
confluence = await multi_timeframe_analyzer.detect_divergence_confluence(
sample_ohlcv_data,
timeframes=["5m", "15m", "1h"],
indicators=["rsi", "macd"]
)
# 결과 검증
assert "confluence_zones" in confluence
assert "divergence_count" in confluence
assert "strength_distribution" in confluence
# 컨플루언스 존 검증
zones = confluence["confluence_zones"]
for zone in zones:
assert "start_index" in zone
assert "end_index" in zone
assert "timeframes_involved" in zone
assert "indicators_involved" in zone
assert "strength" in zone
assert zone["strength"] > 0
@pytest.mark.asyncio
async def test_optimal_timeframe_selection(self, multi_timeframe_analyzer, sample_ohlcv_data):
"""최적 타임프레임 선택 테스트"""
# 최적 타임프레임 선택
optimal_tf = await multi_timeframe_analyzer.select_optimal_timeframe(
sample_ohlcv_data,
trading_style="scalping", # scalping, day_trading, swing_trading
market_conditions={"volatility": "high", "trend": "strong"}
)
# 결과 검증
assert "recommended_timeframe" in optimal_tf
assert "reasoning" in optimal_tf
assert "alternative_timeframes" in optimal_tf
assert "confidence_score" in optimal_tf
# 추천 타임프레임 검증
assert optimal_tf["recommended_timeframe"] in ["1m", "5m", "15m", "1h", "4h", "1d"]
assert 0 <= optimal_tf["confidence_score"] <= 1
# 대안 타임프레임 검증
alternatives = optimal_tf["alternative_timeframes"]
assert len(alternatives) > 0
for alt in alternatives:
assert "timeframe" in alt
assert "score" in alt
@pytest.mark.asyncio
async def test_fractal_analysis(self, multi_timeframe_analyzer, sample_ohlcv_data):
"""프랙탈 분석 테스트"""
# 프랙탈 패턴 분석
fractal_analysis = await multi_timeframe_analyzer.analyze_fractals(
sample_ohlcv_data,
timeframes=["5m", "15m", "1h"],
fractal_period=5
)
# 결과 검증
assert "fractals" in fractal_analysis
assert "fractal_dimensions" in fractal_analysis
assert "self_similarity_score" in fractal_analysis
# 프랙탈 검증
fractals = fractal_analysis["fractals"]
for tf, tf_fractals in fractals.items():
for fractal in tf_fractals:
assert "type" in fractal # up or down
assert "index" in fractal
assert "price" in fractal
assert "strength" in fractal
@pytest.mark.asyncio
async def test_timeframe_transition_analysis(self, multi_timeframe_analyzer, sample_ohlcv_data):
"""타임프레임 전환 분석 테스트"""
# 타임프레임 전환점 분석
transition_analysis = await multi_timeframe_analyzer.analyze_timeframe_transitions(
sample_ohlcv_data,
from_timeframe="5m",
to_timeframe="15m"
)
# 결과 검증
assert "transition_points" in transition_analysis
assert "signal_quality" in transition_analysis
assert "confirmation_rate" in transition_analysis
# 전환점 검증
transitions = transition_analysis["transition_points"]
for point in transitions:
assert "index_5m" in point
assert "index_15m" in point
assert "signal_type" in point
assert "confirmed" in point
assert isinstance(point["confirmed"], bool)
@pytest.mark.asyncio
async def test_volatility_across_timeframes(self, multi_timeframe_analyzer, sample_ohlcv_data):
"""타임프레임별 변동성 분석 테스트"""
# 변동성 분석
volatility_analysis = await multi_timeframe_analyzer.analyze_volatility_profile(
sample_ohlcv_data,
timeframes=["5m", "15m", "1h"],
volatility_window=20
)
# 결과 검증
assert "volatility_by_timeframe" in volatility_analysis
assert "volatility_ratio" in volatility_analysis
assert "expansion_contraction" in volatility_analysis
# 타임프레임별 변동성
vol_by_tf = volatility_analysis["volatility_by_timeframe"]
for tf in ["5m", "15m", "1h"]:
assert tf in vol_by_tf
assert "average" in vol_by_tf[tf]
assert "current" in vol_by_tf[tf]
assert "percentile" in vol_by_tf[tf]
assert 0 <= vol_by_tf[tf]["percentile"] <= 100
@pytest.mark.asyncio
async def test_entry_exit_optimization(self, multi_timeframe_analyzer, sample_ohlcv_data):
"""진입/청산 최적화 테스트"""
# 진입/청산 포인트 최적화
optimization_results = await multi_timeframe_analyzer.optimize_entry_exit_points(
sample_ohlcv_data,
primary_timeframe="15m",
confirmation_timeframes=["5m", "1h"],
strategy_type="momentum"
)
# 결과 검증
assert "entry_points" in optimization_results
assert "exit_points" in optimization_results
assert "risk_reward_ratio" in optimization_results
assert "win_rate_estimate" in optimization_results
# 진입점 검증
entry_points = optimization_results["entry_points"]
for entry in entry_points:
assert "index" in entry
assert "price" in entry
assert "confidence" in entry
assert "confirmations" in entry
assert len(entry["confirmations"]) > 0
@pytest.mark.asyncio
async def test_data_synchronization(self, multi_timeframe_analyzer):
"""데이터 동기화 테스트"""
# 비동기 데이터 생성
data_1m = []
data_5m = []
base_time = datetime(2024, 1, 1, 0, 0, 0)
# 1분 데이터 (정확한 시간)
for i in range(60):
data_1m.append({
"timestamp": (base_time + timedelta(minutes=i)).isoformat(),
"close": 50000 + i * 10
})
# 5분 데이터 (약간 어긋난 시간)
for i in range(12):
# 2초 늦은 타임스탬프
timestamp = base_time + timedelta(minutes=i*5, seconds=2)
data_5m.append({
"timestamp": timestamp.isoformat(),
"close": 50000 + i * 50
})
# 데이터 동기화
synced_data = await multi_timeframe_analyzer.synchronize_timeframe_data({
"1m": data_1m,
"5m": data_5m
})
# 결과 검증
assert "synchronized_timestamps" in synced_data
assert "aligned_data" in synced_data
assert "alignment_quality" in synced_data
# 정렬 품질 확인
assert synced_data["alignment_quality"] > 0.8
@pytest.mark.asyncio
async def test_error_handling(self, multi_timeframe_analyzer):
"""오류 처리 테스트"""
# 잘못된 타임프레임
with pytest.raises(TimeframeError):
await multi_timeframe_analyzer.aggregate_to_timeframe(
[],
target_timeframe="invalid"
)
# 데이터 부족
with pytest.raises(InsufficientDataError):
await multi_timeframe_analyzer.analyze_timeframes(
[{"timestamp": "2024-01-01", "close": 100}], # 1개 데이터
timeframes=["1h", "4h"]
)
# 분석 오류
with pytest.raises(AnalysisError):
await multi_timeframe_analyzer.detect_trend_alignment(
None, # None 데이터
timeframes=["5m"]
)
@pytest.mark.asyncio
async def test_performance_metrics(self, multi_timeframe_analyzer, sample_ohlcv_data):
"""성능 메트릭 테스트"""
# 여러 분석 수행
await multi_timeframe_analyzer.analyze_timeframes(
sample_ohlcv_data,
timeframes=["5m", "15m"]
)
await multi_timeframe_analyzer.detect_trend_alignment(
sample_ohlcv_data,
timeframes=["5m", "15m"]
)
# 성능 메트릭 확인
metrics = multi_timeframe_analyzer.get_performance_metrics()
assert "analysis_count" in metrics
assert "average_processing_time" in metrics
assert "cache_hit_rate" in metrics
assert metrics["analysis_count"] > 0
assert metrics["average_processing_time"] > 0
assert 0 <= metrics["cache_hit_rate"] <= 1
