"""시장 이벤트 플레이북 유닛 테스트.""" import sys import os from datetime import date, timedelta from unittest.mock import patch import pytest # backend를 sys.path에 추가 sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..")) from app.services.market_playbook import ( MarketEventType, EventPhase, PlaybookAction, get_upcoming_events, get_today_playbook, check_realtime_events, trigger_circuit_breaker, clear_circuit_breaker, trigger_vi, is_vi_active, is_circuit_breaker_active, apply_playbook_to_regime, is_daytrading_allowed_by_playbook, is_entry_allowed_by_playbook, get_position_multiplier_from_playbook, _get_nth_weekday_of_month, _get_kr_options_expiry_dates, _get_quad_witching_dates, _get_us_employment_dates, _classify_phase, _is_earnings_season, _is_window_dressing_period, _is_tax_loss_selling_period, _realtime_state, _playbook_cache, ) class TestNthWeekday: """n번째 요일 계산 테스트.""" def test_second_thursday_jan_2026(self): # 2026-01-01은 목요일 -> 둘째 목요일 = 01-08 d = _get_nth_weekday_of_month(2026, 1, 3, 2) assert d == date(2026, 1, 8) def test_third_friday_march_2026(self): # 쿼드러플 위칭: 2026-03 셋째 금요일 d = _get_nth_weekday_of_month(2026, 3, 4, 3) assert d.weekday() == 4 # 금요일 assert 15 <= d.day <= 21 # 셋째 주 def test_first_friday_jan_2026(self): d = _get_nth_weekday_of_month(2026, 1, 4, 1) assert d.weekday() == 4 assert d.day <= 7 class TestRuleBasedDates: """규칙 기반 날짜 계산 테스트.""" def test_kr_options_expiry_12_months(self): dates = _get_kr_options_expiry_dates(2026) assert len(dates) == 12 for d in dates: assert d.weekday() == 3 # 모두 목요일 def test_quad_witching_4_dates(self): dates = _get_quad_witching_dates(2026) assert len(dates) == 4 for d in dates: assert d.weekday() == 4 # 모두 금요일 assert d.month in (3, 6, 9, 12) def test_us_employment_12_months(self): dates = _get_us_employment_dates(2026) assert len(dates) == 12 for d in dates: assert d.weekday() == 4 # 모두 금요일 assert d.day <= 7 # 첫째 주 class TestPhaseClassification: """이벤트 단계 분류 테스트.""" def test_day_of(self): today = date(2026, 1, 28) assert _classify_phase(today, date(2026, 1, 28)) == EventPhase.DAY_OF def test_pre_d1(self): today = date(2026, 1, 27) assert _classify_phase(today, date(2026, 1, 28), pre_days=2) == EventPhase.PRE def test_pre_d2(self): today = date(2026, 1, 26) assert _classify_phase(today, date(2026, 1, 28), pre_days=2) == EventPhase.PRE def test_post_d1(self): today = date(2026, 1, 29) assert _classify_phase(today, date(2026, 1, 28), post_days=1) == EventPhase.POST def test_none_too_far(self): today = date(2026, 1, 20) assert _classify_phase(today, date(2026, 1, 28), pre_days=2) == EventPhase.NONE class TestSeasonalEvents: """시즌 이벤트 판별 테스트.""" def test_earnings_season_jan(self): assert _is_earnings_season(date(2026, 1, 15)) is True def test_earnings_season_jan_early(self): assert _is_earnings_season(date(2026, 1, 5)) is False def test_earnings_season_feb(self): assert _is_earnings_season(date(2026, 2, 15)) is False def test_window_dressing_march(self): assert _is_window_dressing_period(date(2026, 3, 26)) is True assert _is_window_dressing_period(date(2026, 3, 20)) is False def test_tax_loss_selling(self): assert _is_tax_loss_selling_period(date(2026, 12, 15)) is True assert _is_tax_loss_selling_period(date(2026, 11, 15)) is False class TestCircuitBreaker: """서킷브레이커 실시간 이벤트 테스트.""" def setup_method(self): _realtime_state.circuit_breaker_active = False _realtime_state.circuit_breaker_triggered_at = None _realtime_state.vi_stocks.clear() _realtime_state.alerts.clear() _playbook_cache["result"] = None _playbook_cache["date"] = None def test_trigger_and_check(self): assert not is_circuit_breaker_active() trigger_circuit_breaker("test") assert is_circuit_breaker_active() def test_clear(self): trigger_circuit_breaker("test") clear_circuit_breaker() assert not is_circuit_breaker_active() def test_auto_expire(self): import time as _time trigger_circuit_breaker("test") # 만료 시간을 과거로 설정 _realtime_state.circuit_breaker_triggered_at = _time.monotonic() - 2000 assert not is_circuit_breaker_active() class TestVI: """VI(변동성 완화장치) 테스트.""" def setup_method(self): _realtime_state.vi_stocks.clear() _realtime_state.alerts.clear() def test_trigger_and_check(self): assert not is_vi_active("005930") trigger_vi("005930") assert is_vi_active("005930") def test_auto_expire(self): import time as _time trigger_vi("005930") # 쿨다운 만료 시뮬레이션 _realtime_state.vi_stocks["005930"] = _time.monotonic() - 700 assert not is_vi_active("005930") class TestPlaybookAction: """PlaybookAction 직렬화 테스트.""" def test_to_dict(self): action = PlaybookAction( position_multiplier=0.5, entry_allowed=False, daytrading_allowed=False, reason="FOMC 당일", ) d = action.to_dict() assert d["position_multiplier"] == 0.5 assert d["entry_allowed"] is False assert d["daytrading_allowed"] is False assert "FOMC" in d["reason"] class TestApplyPlaybookToRegime: """regime 결과에 플레이북 오버레이 테스트.""" def setup_method(self): _realtime_state.circuit_breaker_active = False _realtime_state.circuit_breaker_triggered_at = None _realtime_state.vi_stocks.clear() _playbook_cache["result"] = None _playbook_cache["date"] = None def test_no_events_passthrough(self): """이벤트 없으면 원본 multiplier 유지.""" regime = {"regime": "bullish", "position_multiplier": 1.0, "max_trades": 5} # 이벤트 없는 날짜로 패치 with patch("app.services.market_playbook.get_today_playbook") as mock_pb: mock_pb.return_value = PlaybookAction( position_multiplier=1.0, entry_allowed=True, daytrading_allowed=True, reason="이벤트 없음", ) result = apply_playbook_to_regime(regime) assert result["position_multiplier"] == 1.0 assert result["regime"] == "bullish" def test_fomc_blocks_entry(self): """FOMC 당일 entry_allowed=False -> regime='event_blocked'.""" regime = {"regime": "bullish", "position_multiplier": 1.0, "max_trades": 5, "min_score": 1} with patch("app.services.market_playbook.get_today_playbook") as mock_pb: mock_pb.return_value = PlaybookAction( position_multiplier=0.0, entry_allowed=False, daytrading_allowed=False, reason="FOMC 당일", ) result = apply_playbook_to_regime(regime) assert result["regime"] == "event_blocked" assert result["max_trades"] == 0 assert result["position_multiplier"] == 0.0 def test_multiplier_stacks(self): """bear(0.5) * playbook(0.5) = 0.25.""" regime = {"regime": "bearish", "position_multiplier": 0.5, "max_trades": 3} with patch("app.services.market_playbook.get_today_playbook") as mock_pb: mock_pb.return_value = PlaybookAction( position_multiplier=0.5, entry_allowed=True, daytrading_allowed=True, reason="옵션 만기 D-1", ) result = apply_playbook_to_regime(regime) assert result["position_multiplier"] == 0.25 def test_circuit_breaker_overrides_all(self): """서킷브레이커 발동 시 모든 매매 차단.""" trigger_circuit_breaker("test") regime = {"regime": "bullish", "position_multiplier": 1.0, "max_trades": 5, "min_score": 1} result = apply_playbook_to_regime(regime) assert result["regime"] == "event_blocked" assert result["position_multiplier"] == 0.0 # 정리 clear_circuit_breaker() class TestHelperFunctions: """통합 헬퍼 함수 테스트.""" def setup_method(self): _realtime_state.circuit_breaker_active = False _realtime_state.vi_stocks.clear() _playbook_cache["result"] = None _playbook_cache["date"] = None def test_is_daytrading_allowed_default(self): """기본 상태에서 데이트레이딩 허용.""" # 이벤트 없는 날이면 True result = is_daytrading_allowed_by_playbook() assert isinstance(result, bool) def test_is_entry_allowed_default(self): result = is_entry_allowed_by_playbook() assert isinstance(result, bool) def test_get_position_multiplier_default(self): result = get_position_multiplier_from_playbook() assert 0.0 <= result <= 1.0 if __name__ == "__main__": pytest.main([__file__, "-v"])