"""리스크 검증 모듈 (risk_validator.py) — MCP 서버용 모든 매매 주문을 실행 전에 CLAUDE.md 섹션 5 리스크 규칙으로 검증한다. 검증 항목: 1. 가격 필터: 국내 5,000원 초과 (ETF/리츠 예외) 2. 포지션 사이징: 종목당 총자본 4% 이내 3. 최대 보유 종목: 20개 이내 4. MDD 상태: -10% 신규 진입 중단, -20% 강제 축소 5. 레버리지 ETF 한도: 총자본 20% 이내 아키텍처: - DB 읽기: AsyncSessionLocal 직접 사용 (SELECT only) - DB 쓰기: 없음 (검증만 수행) - 싱글톤: risk_validator 인스턴스, asyncio.Lock으로 동시성 보호 - 공개 API: validate_and_log_order() — 종합 검증 + 로깅 [주의] 이 모듈은 자체 DB 엔진을 생성한다 (백엔드 database.py와 별도). MCP 서버가 백엔드와 다른 프로세스에서 실행되기 때문. """ import asyncio import logging import sys import os from typing import Optional, Dict, Tuple from datetime import datetime, timezone from decimal import Decimal # Backend imports sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '../backend'))) from sqlalchemy import select from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession from sqlalchemy.orm import sessionmaker from app.core.config import settings from app.models.models import Portfolio, ActiveTrade, TradeHistory from app.services.korea_investment_api import korea_investment_api logger = logging.getLogger(__name__) # Backend DB connection engine = create_async_engine(settings.DATABASE_URL, echo=False) AsyncSessionLocal = sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) # Risk limits from CLAUDE.md Section 5 MAX_POSITION_SIZE_PCT = settings.MAX_POSITION_SIZE # config.py와 동기화 (백테스트 검증 완료: 0.06) MAX_HOLDINGS_COUNT = 20 # CLAUDE.md 5.1: max 20 holdings MIN_DOMESTIC_PRICE = 5000 # KRW MAX_LEVERAGE_ETF_PCT = 0.20 # max 20% of capital for leverage/inverse ETF # Leverage/Inverse ETF identifiers LEVERAGE_ETF_KEYWORDS = ['레버리지', 'LEVERAGE', '2X', '3X', 'INVERSE', '인버스', '-1X', '-2X', '-3X'] class RiskValidationResult: """Result of a risk validation check""" def __init__(self, allowed: bool, reason: str = "", details: Optional[Dict] = None): self.allowed = allowed self.reason = reason self.details = details or {} self.timestamp = datetime.now(timezone.utc).isoformat() def __repr__(self): return f"RiskValidationResult(allowed={self.allowed}, reason='{self.reason}')" class RiskValidator: """Risk validation engine for Claude-based auto trading""" def __init__(self): self.db_session = None self._lock = asyncio.Lock() async def _get_session(self) -> AsyncSession: """Get or create DB session. DEPRECATED: 새 코드에서는 _session_ctx()를 사용하세요. 기존 시그니처 유지를 위해 남겨두되, 내부적으로 새 세션을 생성합니다. """ if self.db_session is None: self.db_session = AsyncSessionLocal() return self.db_session async def _close_session(self): """Close DB session""" if self.db_session is not None: try: await self.db_session.close() except Exception: pass self.db_session = None def _session_ctx(self): """컨텍스트 매니저로 세션을 안전하게 관리. 사용법: async with self._session_ctx() as session: result = await session.execute(...) """ return AsyncSessionLocal() async def validate_price_filter(self, stock_code: str, market: str = "KRX") -> RiskValidationResult: """ Check price filter: min 5,000 KRW for domestic stocks Args: stock_code: Stock code to validate market: Market code (default: KRX for domestic) Returns: RiskValidationResult """ try: # Skip price check for overseas/ETF markets if market and market not in ["KRX", "KOSPI", "KOSDAQ"]: logger.info(f"[{stock_code}] Price filter skipped (overseas market: {market})") return RiskValidationResult(allowed=True) # Get current price from KIS API (공개 메서드 사용) # get_current_price()는 dict를 반환: {"output": {"stck_prpr": "50000", ...}} price_info = await korea_investment_api.get_current_price(stock_code) current_price = int(price_info.get('output', {}).get('stck_prpr', 0)) if not current_price or current_price <= 0: return RiskValidationResult( allowed=False, reason=f"Failed to fetch current price for {stock_code}" ) if current_price <= MIN_DOMESTIC_PRICE: return RiskValidationResult( allowed=False, reason=f"Price filter violation: {current_price} KRW < {MIN_DOMESTIC_PRICE} KRW minimum", details={"current_price": current_price, "min_price": MIN_DOMESTIC_PRICE} ) logger.info(f"[{stock_code}] Price filter passed: {current_price} KRW") return RiskValidationResult( allowed=True, details={"current_price": current_price} ) except Exception as e: logger.error(f"Price filter check error for {stock_code}: {e}") return RiskValidationResult(allowed=False, reason=f"Price check error: {str(e)}") async def validate_position_size(self, stock_code: str, order_quantity: int, entry_price: float, account_balance: float) -> RiskValidationResult: """ Check position sizing: max 3-4% per stock Args: stock_code: Stock code to validate order_quantity: Number of shares to buy entry_price: Entry price per share account_balance: Current account balance Returns: RiskValidationResult """ try: order_value = order_quantity * entry_price position_pct = (order_value / account_balance) * 100 if account_balance > 0 else 0 if position_pct > MAX_POSITION_SIZE_PCT * 100: return RiskValidationResult( allowed=False, reason=f"Position size violation: {position_pct:.2f}% > {MAX_POSITION_SIZE_PCT*100}% maximum", details={ "order_value": order_value, "position_pct": position_pct, "max_allowed_pct": MAX_POSITION_SIZE_PCT * 100 } ) logger.info(f"[{stock_code}] Position size check passed: {position_pct:.2f}% of account") return RiskValidationResult( allowed=True, details={"position_pct": position_pct, "order_value": order_value} ) except Exception as e: logger.error(f"Position size check error for {stock_code}: {e}") return RiskValidationResult(allowed=False, reason=f"Position size check error: {str(e)}") async def validate_max_holdings(self, account_balance: float) -> RiskValidationResult: """ Check max holdings limit: 20-25 concurrent positions Args: account_balance: Current account balance (for context) Returns: RiskValidationResult """ try: async with self._session_ctx() as db: # Count active holdings (quantity > 0) result = await db.execute( select(Portfolio).where(Portfolio.quantity > 0) ) active_holdings = result.scalars().all() holdings_count = len(active_holdings) if holdings_count >= MAX_HOLDINGS_COUNT: return RiskValidationResult( allowed=False, reason=f"Max holdings limit reached: {holdings_count} >= {MAX_HOLDINGS_COUNT}", details={"current_holdings": holdings_count, "max_allowed": MAX_HOLDINGS_COUNT} ) logger.info(f"Max holdings check passed: {holdings_count}/{MAX_HOLDINGS_COUNT}") return RiskValidationResult( allowed=True, details={"current_holdings": holdings_count, "available_slots": MAX_HOLDINGS_COUNT - holdings_count} ) except Exception as e: logger.error(f"Max holdings check error: {e}") return RiskValidationResult(allowed=False, reason=f"Max holdings check error: {str(e)}") async def validate_mdd_state(self) -> RiskValidationResult: """ Check MDD (Maximum Drawdown) state: - MDD -10%: pause new entries - MDD -20%: force reduce positions Returns: RiskValidationResult """ try: # Import MDD manager from app.services.mdd_manager import mdd_manager # Check if new positions are paused due to MDD can_open = await mdd_manager.can_open_new_position() if not can_open: state = { "is_paused": mdd_manager.is_paused, "reduction_applied": mdd_manager.reduction_applied, "peak_balance": mdd_manager.peak_balance } return RiskValidationResult( allowed=False, reason=f"MDD limit triggered (reduction: {mdd_manager.reduction_applied}). New entries paused.", details=state ) logger.info(f"MDD check passed: trading active (peak: {mdd_manager.peak_balance})") return RiskValidationResult( allowed=True, details={ "is_paused": mdd_manager.is_paused, "reduction_applied": mdd_manager.reduction_applied, "peak_balance": mdd_manager.peak_balance } ) except Exception as e: logger.error(f"MDD check error: {e}") # Don't block trading if MDD check fails - log and allow logger.warning(f"MDD validation skipped due to error: {str(e)}") return RiskValidationResult(allowed=True, reason="MDD check skipped (error)") async def validate_leverage_etf_limit(self, stock_code: str, stock_name: str, order_quantity: int, entry_price: float, account_balance: float) -> RiskValidationResult: """ Check leverage/inverse ETF limit: max 20% of capital Args: stock_code: Stock code stock_name: Stock name (for keyword matching) order_quantity: Number of shares to buy entry_price: Entry price per share account_balance: Current account balance Returns: RiskValidationResult """ try: # Check if this is a leverage/inverse ETF is_leverage_etf = any(keyword in stock_name.upper() for keyword in LEVERAGE_ETF_KEYWORDS) if not is_leverage_etf: logger.info(f"[{stock_code}] Not a leverage ETF, skipping leverage limit check") return RiskValidationResult(allowed=True) # Calculate total leverage ETF position async with self._session_ctx() as db: result = await db.execute(select(Portfolio)) all_holdings = result.scalars().all() leverage_etf_value = 0 for holding in all_holdings: if any(keyword in (holding.stock_name or "").upper() for keyword in LEVERAGE_ETF_KEYWORDS): holding_value = float(holding.current_price or 0) * holding.quantity leverage_etf_value += holding_value # Add proposed order proposed_value = order_quantity * entry_price total_leverage_value = leverage_etf_value + proposed_value leverage_pct = (total_leverage_value / account_balance) * 100 if account_balance > 0 else 0 if leverage_pct > MAX_LEVERAGE_ETF_PCT * 100: return RiskValidationResult( allowed=False, reason=f"Leverage ETF limit: {leverage_pct:.2f}% > {MAX_LEVERAGE_ETF_PCT*100}% maximum", details={ "current_leverage_value": leverage_etf_value, "proposed_value": proposed_value, "total_leverage_pct": leverage_pct, "max_allowed_pct": MAX_LEVERAGE_ETF_PCT * 100 } ) logger.info(f"[{stock_code}] Leverage ETF limit check passed: {leverage_pct:.2f}% of account") return RiskValidationResult( allowed=True, details={"leverage_etf_pct": leverage_pct, "total_leverage_value": total_leverage_value} ) except Exception as e: logger.error(f"Leverage ETF limit check error for {stock_code}: {e}") return RiskValidationResult(allowed=False, reason=f"Leverage ETF check error: {str(e)}") async def validate_order_comprehensive(self, stock_code: str, stock_name: str, order_quantity: int, entry_price: float, account_balance: float, market: str = "KRX") -> Tuple[bool, Dict]: """ Comprehensive order validation combining all risk checks Args: stock_code: Stock code stock_name: Stock name order_quantity: Number of shares entry_price: Entry price per share account_balance: Current account balance market: Market code (default: KRX) Returns: Tuple of (allowed: bool, details: Dict with all checks) """ try: async with self._lock: validation_results = { "timestamp": datetime.now(timezone.utc).isoformat(), "stock_code": stock_code, "stock_name": stock_name, "order_quantity": order_quantity, "entry_price": entry_price, "account_balance": account_balance, "checks": {} } # Run all checks checks = [ ("price_filter", await self.validate_price_filter(stock_code, market)), ("position_size", await self.validate_position_size(stock_code, order_quantity, entry_price, account_balance)), ("max_holdings", await self.validate_max_holdings(account_balance)), ("mdd_state", await self.validate_mdd_state()), ("leverage_etf_limit", await self.validate_leverage_etf_limit(stock_code, stock_name, order_quantity, entry_price, account_balance)) ] # Record all check results all_passed = True for check_name, result in checks: validation_results["checks"][check_name] = { "allowed": result.allowed, "reason": result.reason, "details": result.details } if not result.allowed: all_passed = False logger.warning(f"❌ [{stock_code}] {check_name}: {result.reason}") else: logger.info(f"✅ [{stock_code}] {check_name}: passed") validation_results["overall_allowed"] = all_passed if all_passed: logger.info(f"✅✅✅ [{stock_code}] ALL RISK CHECKS PASSED - Order approved") else: logger.warning(f"❌ [{stock_code}] RISK VALIDATION FAILED - Order rejected") return all_passed, validation_results except Exception as e: logger.error(f"Comprehensive validation error for {stock_code}: {e}", exc_info=True) return False, { "timestamp": datetime.now(timezone.utc).isoformat(), "stock_code": stock_code, "overall_allowed": False, "error": str(e) } # Singleton instance risk_validator = RiskValidator() async def validate_and_log_order(stock_code: str, stock_name: str, order_quantity: int, entry_price: float, account_balance: float, market: str = "KRX") -> Tuple[bool, Dict]: """ Public API for order validation with comprehensive logging Args: stock_code: Stock code stock_name: Stock name order_quantity: Number of shares entry_price: Entry price per share account_balance: Current account balance market: Market code Returns: Tuple of (allowed: bool, validation_details: Dict) """ allowed, details = await risk_validator.validate_order_comprehensive( stock_code=stock_code, stock_name=stock_name, order_quantity=order_quantity, entry_price=entry_price, account_balance=account_balance, market=market ) # Log rejection with full details if not allowed: failed_checks = [ f"{check}: {data['reason']}" for check, data in details.get('checks', {}).items() if not data.get('allowed', True) ] logger.error(f"🚫 ORDER REJECTED [{stock_code}]: {'; '.join(failed_checks)}") return allowed, details