"""미장 심리 분석 + 한국 시장 레짐 보정 (07:00 KST 프리마켓 배치) [목적] 전날 밤 미국 시장(S&P500, NASDAQ, VIX) 종가를 기반으로 한국 장 개장 전 적응형 전략 모드를 보정한다. [데이터 소스] 1순위: 네이버 금융 세계지수 (S&P500, NASDAQ) 2순위: Yahoo Finance API (VIX) 폴백: KIS 해외주식 현재가 (SPY, QQQ) [보정 규칙] - VIX >= 30: Defensive 모드 강제 (위기 접근) - VIX >= 35: Crisis 모드 강제 - S&P500 전일 -2% 이상 하락: Defensive 이상으로 상향 - S&P500 전일 -3% 이상 하락: Crisis 강제 - 미장 정상 (VIX < 25, S&P +/-1% 이내): 보정 없음 [출력] Redis 캐시: us_market_sentiment:latest (TTL 4시간) 적응형 전략 모드 오버라이드 (필요 시) """ import asyncio import logging from dataclasses import dataclass, asdict from datetime import datetime from typing import Optional, Dict import pytz logger = logging.getLogger(__name__) _KST = pytz.timezone("Asia/Seoul") @dataclass class USMarketSentiment: """미장 심리 분석 결과.""" # S&P500 sp500_price: float = 0.0 sp500_change_pct: float = 0.0 # NASDAQ nasdaq_price: float = 0.0 nasdaq_change_pct: float = 0.0 # DOW dow_price: float = 0.0 dow_change_pct: float = 0.0 # VIX vix_price: float = 20.0 vix_change_pct: float = 0.0 # 종합 판단 sentiment: str = "NEUTRAL" # RISK_ON / NEUTRAL / RISK_OFF / PANIC regime_override: str = "" # 빈 문자열이면 오버라이드 없음 description: str = "" fetched_at: str = "" data_source: str = "naver+yahoo" async def fetch_us_market_data() -> USMarketSentiment: """미장 데이터 수집 (네이버 세계지수 + Yahoo VIX).""" from app.services.naver_finance import get_world_indices, get_vix_index result = USMarketSentiment() result.fetched_at = datetime.now(_KST).isoformat() # 1. 네이버 세계지수 (S&P500, NASDAQ, DOW) try: indices = await get_world_indices() if "SP500" in indices: result.sp500_price = indices["SP500"]["price"] result.sp500_change_pct = indices["SP500"]["change_pct"] if "NASDAQ" in indices: result.nasdaq_price = indices["NASDAQ"]["price"] result.nasdaq_change_pct = indices["NASDAQ"]["change_pct"] if "DOW" in indices: result.dow_price = indices["DOW"]["price"] result.dow_change_pct = indices["DOW"]["change_pct"] except Exception as e: logger.warning("[US_SENTIMENT] 네이버 세계지수 조회 실패: %s", e) # 2. VIX (Yahoo Finance) try: vix = await get_vix_index() if vix: result.vix_price = vix["price"] result.vix_change_pct = vix["change_pct"] except Exception as e: logger.warning("[US_SENTIMENT] VIX 조회 실패: %s", e) # 3. 폴백: KIS API (네이버/Yahoo 모두 실패 시) if result.sp500_price == 0.0: try: from app.services.korea_investment_api import korea_investment_api spy_data = await korea_investment_api.get_overseas_price("SPY", "NASD") output = spy_data.get("output", {}) result.sp500_price = float(output.get("last", 0)) prev = float(output.get("base", 0)) if prev > 0: result.sp500_change_pct = round((result.sp500_price - prev) / prev * 100, 2) result.data_source = "kis_fallback" logger.info("[US_SENTIMENT] KIS 폴백으로 SPY 조회: %.2f", result.sp500_price) except Exception as e: logger.warning("[US_SENTIMENT] KIS 폴백도 실패: %s", e) # 4. 종합 심리 판단 result.sentiment, result.regime_override, result.description = _analyze_sentiment(result) return result def _analyze_sentiment(data: USMarketSentiment) -> tuple: """미장 데이터 기반 심리 판단 + 레짐 오버라이드 결정. Returns: (sentiment, regime_override, description) """ vix = data.vix_price sp500_chg = data.sp500_change_pct nasdaq_chg = data.nasdaq_change_pct # PANIC: VIX >= 35 또는 S&P500 -3% 이상 하락 if vix >= 35 or sp500_chg <= -3.0: return ( "PANIC", "crisis", f"미장 패닉 (VIX={vix:.1f}, S&P500={sp500_chg:+.1f}%, NASDAQ={nasdaq_chg:+.1f}%)", ) # RISK_OFF: VIX >= 30 또는 S&P500 -2% 이상 하락 if vix >= 30 or sp500_chg <= -2.0: return ( "RISK_OFF", "defensive", f"미장 리스크오프 (VIX={vix:.1f}, S&P500={sp500_chg:+.1f}%, NASDAQ={nasdaq_chg:+.1f}%)", ) # RISK_ON: VIX < 18 이고 S&P500 +1% 이상 상승 if vix < 18 and sp500_chg >= 1.0: return ( "RISK_ON", "", # 오버라이드 없음 (공격적 전환은 국내 데이터 기반으로) f"미장 리스크온 (VIX={vix:.1f}, S&P500={sp500_chg:+.1f}%, NASDAQ={nasdaq_chg:+.1f}%)", ) # NEUTRAL: 기본 return ( "NEUTRAL", "", f"미장 중립 (VIX={vix:.1f}, S&P500={sp500_chg:+.1f}%, NASDAQ={nasdaq_chg:+.1f}%)", ) async def run_pre_market_check() -> Dict: """07:00 KST 프리마켓 배치 메인 함수. 1. 미장 데이터 수집 2. 심리 분석 3. Redis 캐시 저장 4. 필요 시 적응형 전략 모드 오버라이드 5. Telegram 알림 """ logger.info("[PRE_MARKET] 미장 심리 분석 시작...") sentiment = await fetch_us_market_data() sentiment_dict = asdict(sentiment) # Redis 캐시 저장 (TTL 4시간 = 14400초) try: from app.core.redis import setex_json await setex_json("us_market_sentiment:latest", 14400, sentiment_dict) logger.info("[PRE_MARKET] Redis 캐시 저장 완료") except Exception as e: logger.warning("[PRE_MARKET] Redis 저장 실패: %s", e) # 적응형 전략 모드 오버라이드 if sentiment.regime_override: try: from app.services.adaptive_strategy import force_mode_override force_mode_override(sentiment.regime_override, reason=sentiment.description) logger.warning( "[PRE_MARKET] 적응형 전략 모드 오버라이드: %s → %s", sentiment.sentiment, sentiment.regime_override, ) except Exception as e: logger.warning("[PRE_MARKET] 모드 오버라이드 실패: %s", e) # Telegram 알림 try: from app.services.telegram_notifier import telegram_notifier msg = ( f"🌍 미장 프리마켓 리포트\n" f"━━━━━━━━━━━━━━━\n" f"S&P500: {sentiment.sp500_price:,.1f} ({sentiment.sp500_change_pct:+.1f}%)\n" f"NASDAQ: {sentiment.nasdaq_price:,.1f} ({sentiment.nasdaq_change_pct:+.1f}%)\n" f"DOW: {sentiment.dow_price:,.1f} ({sentiment.dow_change_pct:+.1f}%)\n" f"VIX: {sentiment.vix_price:.1f} ({sentiment.vix_change_pct:+.1f}%)\n" f"━━━━━━━━━━━━━━━\n" f"판단: {sentiment.sentiment}\n" ) if sentiment.regime_override: msg += f"⚠️ 전략 오버라이드: {sentiment.regime_override}\n" msg += f"📝 {sentiment.description}" asyncio.ensure_future(telegram_notifier.send_message(msg)) except Exception: pass logger.info( "[PRE_MARKET] 완료 — %s (VIX=%.1f, S&P500=%+.1f%%, override=%s)", sentiment.sentiment, sentiment.vix_price, sentiment.sp500_change_pct, sentiment.regime_override or "없음", ) return sentiment_dict