"""종목별 패턴 메모리 시스템 유닛 테스트. 순수 로직 함수만 테스트 (DB/Redis 비의존). """ import pytest from app.services.stock_pattern_memory import ( StockProfile, analyze_support_resistance, analyze_rsi_bounce_history, analyze_volume_pattern, calculate_stock_beta, get_dynamic_rsi_threshold, get_dynamic_stop_loss, is_near_support, is_near_resistance, get_entry_confidence_boost, _calculate_rsi_series, _calculate_daily_range_pct, ) # ═════════════════════════════════════════════════════════════════════════════ # StockProfile 직렬화 # ═════════════════════════════════════════════════════════════════════════════ class TestStockProfile: def test_to_dict_and_from_dict_roundtrip(self): profile = StockProfile( stock_code="005930", stock_name="삼성전자", support_levels=[58000.0, 60000.0], resistance_levels=[75000.0, 80000.0], win_rate=0.65, total_trades=20, ) d = profile.to_dict() restored = StockProfile.from_dict(d) assert restored.stock_code == "005930" assert restored.support_levels == [58000.0, 60000.0] assert restored.win_rate == 0.65 def test_from_dict_ignores_unknown_keys(self): data = {"stock_code": "005930", "unknown_field": 42} profile = StockProfile.from_dict(data) assert profile.stock_code == "005930" # ═════════════════════════════════════════════════════════════════════════════ # 지지/저항선 분석 # ═════════════════════════════════════════════════════════════════════════════ class TestSupportResistance: def _generate_sine_data(self, n=120, base=50000, amplitude=5000): """지지/저항선이 명확한 사인파 데이터 생성.""" import math closes = [base + amplitude * math.sin(2 * math.pi * i / 20) for i in range(n)] highs = [c + 200 for c in closes] lows = [c - 200 for c in closes] return closes, highs, lows def test_detects_levels_from_sine_wave(self): closes, highs, lows = self._generate_sine_data() supports, resistances = analyze_support_resistance(closes, highs, lows, window=100) assert len(supports) > 0, "지지선이 하나 이상 탐지되어야 한다" assert len(resistances) > 0, "저항선이 하나 이상 탐지되어야 한다" def test_returns_empty_on_insufficient_data(self): supports, resistances = analyze_support_resistance([1, 2, 3], [2, 3, 4], [0, 1, 2]) assert supports == [] assert resistances == [] def test_max_five_levels(self): """최대 5개 레벨만 반환.""" closes, highs, lows = self._generate_sine_data(n=300) supports, resistances = analyze_support_resistance(closes, highs, lows, window=200) assert len(supports) <= 5 assert len(resistances) <= 5 # ═════════════════════════════════════════════════════════════════════════════ # RSI 반등 통계 # ═════════════════════════════════════════════════════════════════════════════ class TestRsiBounceHistory: def test_default_on_insufficient_data(self): avg, mn, top = analyze_rsi_bounce_history([100, 101, 102]) assert avg == 30.0 assert mn == 20.0 assert top == 70.0 def test_finds_bounce_points(self): """충분한 데이터로 반등 통계 산출.""" import math # RSI가 진동하도록 가격 생성 (120일) closes = [50000 + 3000 * math.sin(2 * math.pi * i / 14) for i in range(120)] avg, mn, top = analyze_rsi_bounce_history(closes) # 기본값이 아닌 값이 산출되어야 한다 assert isinstance(avg, float) assert isinstance(mn, float) assert isinstance(top, float) # ═════════════════════════════════════════════════════════════════════════════ # 거래량 패턴 # ═════════════════════════════════════════════════════════════════════════════ class TestVolumePattern: def test_default_on_insufficient_data(self): threshold, avg = analyze_volume_pattern([100, 200]) assert threshold == 2.0 assert avg == 0.0 def test_spike_threshold_calculation(self): # 일반 거래량 + 간헐적 급증 volumes = [1000] * 50 + [5000, 6000, 7000] # 53일 threshold, avg = analyze_volume_pattern(volumes) assert threshold >= 1.5 assert threshold <= 5.0 assert avg > 0 def test_clamp_range(self): # 거래량이 균일하면 spike_threshold가 낮게 나옴 → 1.5 하한 적용 volumes = [1000] * 60 threshold, avg = analyze_volume_pattern(volumes) assert threshold >= 1.5 # ═════════════════════════════════════════════════════════════════════════════ # 베타 계산 # ═════════════════════════════════════════════════════════════════════════════ class TestBeta: def test_identical_series_beta_one(self): """동일한 수익률이면 베타 = 1.0.""" closes = [100 + i for i in range(60)] beta = calculate_stock_beta(closes, closes) assert abs(beta - 1.0) < 0.1 def test_insufficient_data(self): beta = calculate_stock_beta([100, 101], [200, 201]) assert beta == 1.0 def test_inverse_beta_negative(self): """역방향 수익률이면 베타 < 0.""" import math # 사인파 사용: 주기가 같고 위상이 반대 stock = [100 + 10 * math.sin(2 * math.pi * i / 20) for i in range(60)] index = [100 - 10 * math.sin(2 * math.pi * i / 20) for i in range(60)] beta = calculate_stock_beta(stock, index) assert beta < 0 # ═════════════════════════════════════════════════════════════════════════════ # 동적 RSI 임계값 # ═════════════════════════════════════════════════════════════════════════════ class TestDynamicRsiThreshold: def test_insufficient_trades_returns_default(self): profile = StockProfile(stock_code="005930", total_trades=1) # total_trades < 3 → 기본값 settings.MEAN_REVERSION_RSI_THRESHOLD (46.0) threshold = get_dynamic_rsi_threshold(profile) assert threshold == 46.0 def test_low_bounce_avg(self): """반등 평균 RSI가 낮은 종목 → 낮은 임계값.""" profile = StockProfile( stock_code="373220", stock_name="에코프로", rsi_bounce_avg=22.0, total_trades=10, ) threshold = get_dynamic_rsi_threshold(profile) assert threshold == 35.0 # 22 + 5 = 27 → 하한 클램프 35 (BEAR_RSI_THRESHOLD) def test_high_bounce_avg(self): """반등 평균 RSI가 높은 종목 → 상한 클램프.""" profile = StockProfile( stock_code="005930", stock_name="삼성전자", rsi_bounce_avg=38.0, total_trades=10, ) threshold = get_dynamic_rsi_threshold(profile) assert threshold == 43.0 # 38 + 5 = 43 → 상한 46이므로 클램프 안 됨 def test_normal_bounce_avg(self): profile = StockProfile( stock_code="035720", stock_name="카카오", rsi_bounce_avg=32.0, total_trades=10, ) threshold = get_dynamic_rsi_threshold(profile) assert threshold == 37.0 # 32 + 5 = 37 # ═════════════════════════════════════════════════════════════════════════════ # 동적 손절 # ═════════════════════════════════════════════════════════════════════════════ class TestDynamicStopLoss: def test_low_volatility(self): profile = StockProfile(stock_code="005930", volatility_percentile=10.0) sl = get_dynamic_stop_loss(profile) assert sl == -5.0 def test_high_volatility(self): profile = StockProfile(stock_code="373220", volatility_percentile=90.0) sl = get_dynamic_stop_loss(profile) assert sl == -10.0 def test_mid_volatility(self): profile = StockProfile(stock_code="035720", volatility_percentile=50.0) sl = get_dynamic_stop_loss(profile) assert -10.0 < sl < -5.0 # ═════════════════════════════════════════════════════════════════════════════ # 지지/저항선 근접 판별 # ═════════════════════════════════════════════════════════════════════════════ class TestNearSupport: def test_near_support(self): profile = StockProfile(stock_code="005930", support_levels=[60000.0, 58000.0]) assert is_near_support(profile, 59500.0) is True # 58000 기준 +2.6% assert is_near_support(profile, 70000.0) is False def test_near_resistance(self): profile = StockProfile(stock_code="005930", resistance_levels=[80000.0]) assert is_near_resistance(profile, 79000.0) is True # -1.25% assert is_near_resistance(profile, 60000.0) is False def test_empty_levels(self): profile = StockProfile(stock_code="005930") assert is_near_support(profile, 50000.0) is False assert is_near_resistance(profile, 50000.0) is False # ═════════════════════════════════════════════════════════════════════════════ # 신뢰도 보정 # ═════════════════════════════════════════════════════════════════════════════ class TestEntryConfidenceBoost: def test_boost_range(self): """결과가 항상 -0.3 ~ +0.3 범위.""" profile = StockProfile( stock_code="005930", support_levels=[60000.0], resistance_levels=[80000.0], rsi_bounce_avg=30.0, rsi_bounce_min=20.0, volume_spike_threshold=2.0, avg_volume_20d=1000000, win_rate=0.7, total_trades=10, ) boost = get_entry_confidence_boost(profile, rsi=28.0, current_price=60500.0, current_volume=3000000) assert -0.3 <= boost <= 0.3 def test_all_positive_signals(self): """지지선 + RSI 반등 + 높은 승률 + 거래량 급증 → 양수 boost.""" profile = StockProfile( stock_code="005930", support_levels=[60000.0], resistance_levels=[90000.0], # 먼 저항선 rsi_bounce_avg=28.0, rsi_bounce_min=20.0, volume_spike_threshold=2.0, avg_volume_20d=1000000, win_rate=0.7, total_trades=10, ) boost = get_entry_confidence_boost(profile, rsi=28.0, current_price=60500.0, current_volume=3000000) assert boost > 0 def test_all_negative_signals(self): """저항선 + 낮은 승률 → 음수 boost.""" profile = StockProfile( stock_code="005930", support_levels=[40000.0], # 먼 지지선 resistance_levels=[60000.0], win_rate=0.2, total_trades=10, avg_volume_20d=1000000, volume_spike_threshold=3.0, ) boost = get_entry_confidence_boost(profile, rsi=50.0, current_price=60500.0, current_volume=500000) assert boost < 0 def test_no_profile_data(self): """기본값 프로필 → boost 0.""" profile = StockProfile(stock_code="005930") boost = get_entry_confidence_boost(profile, rsi=35.0, current_price=70000.0, current_volume=0) assert boost == 0.0 # ═════════════════════════════════════════════════════════════════════════════ # RSI 시리즈 계산 # ═════════════════════════════════════════════════════════════════════════════ class TestRsiSeries: def test_length(self): closes = [100 + i * 0.5 for i in range(50)] series = _calculate_rsi_series(closes, 14) assert len(series) > 0 assert len(series) == len(closes) - 14 - 1 # deltas = n-1, skip first period def test_monotonic_up_high_rsi(self): """단조 증가 → RSI > 50.""" closes = list(range(100, 200)) series = _calculate_rsi_series(closes, 14) for val in series: assert val > 50 def test_insufficient_data(self): series = _calculate_rsi_series([100, 101, 102], 14) assert series == [] # ═════════════════════════════════════════════════════════════════════════════ # 일일 변동폭 # ═════════════════════════════════════════════════════════════════════════════ class TestDailyRange: def test_basic_calculation(self): highs = [105.0] * 20 lows = [95.0] * 20 closes = [100.0] * 20 result = _calculate_daily_range_pct(highs, lows, closes) assert abs(result - 10.0) < 0.1 # (105-95)/100 = 10% def test_insufficient_data(self): result = _calculate_daily_range_pct([105], [95], [100]) assert result == 0.0