MCP Market Statistics Server

"""멀티 타임프레임 분석기 테스트""" 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