""" 모델 검증 시스템 (Market-Regime Aware Model Validator) - 백테스트 기반 현재 LSTM v3 성능 검증 - 코스피 레짐별 모델 적합성 평가 - 코스피 6300 강세장 시나리오 대응 점검 - 모델 교체 권고 보고서 생성 사용법: validator = ModelValidator() report = validator.validate(ticker, ohlcv_data, strategy_fn, kospi_price=2600) print(report.summary()) validator.send_alert(report) # 텔레그램 알림 (심각한 경우만) """ import os import json import time from dataclasses import dataclass, field from typing import Optional, List from modules.config import Config from modules.analysis.backtest import Backtester, BacktestResult from modules.analysis.market_regime import MarketRegimeDetector, MarketRegime, RegimeAnalysis # 모델 적합성 최소 기준 _MIN_SHARPE = 0.5 _MIN_WIN_RATE = 50.0 # % _MAX_MDD = -20.0 # % (초과 시 문제) _MIN_PROFIT_FACTOR = 1.2 _CACHE_TTL_SECONDS = 86400 # 24시간 @dataclass class ValidationReport: """모델 검증 보고서""" ticker: str kospi_level: float regime: str regime_description: str backtest_result: Optional[BacktestResult] model_suitable: bool suitability_score: float # 0~1 issues: List[str] = field(default_factory=list) recommendations: List[str] = field(default_factory=list) alternative_models: List[str] = field(default_factory=list) regime_strategy_hint: str = "" risk_level: str = "LOW" def summary(self) -> str: lines = [ "=" * 55, f"🔍 모델 검증 보고서 [{self.ticker}]", "=" * 55, f"코스피 수준 : {self.kospi_level:.0f} ({self.regime_description})", f"시장 레짐 : {self.regime} [리스크: {self.risk_level}]", f"모델 적합성 : {'✅ 적합' if self.model_suitable else '⚠️ 부적합'} " f"({self.suitability_score:.0%})", ] if self.backtest_result: bt = self.backtest_result lines += [ "", "📊 백테스트 성과", f" 총 수익률 : {bt.total_return_pct:+.2f}%", f" Sharpe Ratio : {bt.sharpe_ratio:.3f}", f" Max Drawdown : {bt.max_drawdown_pct:.2f}%", f" 승률 : {bt.win_rate:.1f}% ({bt.winning_trades}/{bt.total_trades})", f" 손익비(PF) : {bt.profit_factor:.2f}", ] if self.issues: lines.append("") lines.append(f"⚠️ 발견된 문제 ({len(self.issues)}건)") for issue in self.issues: lines.append(f" - {issue}") if self.recommendations: lines.append("") lines.append("💡 권고사항") for rec in self.recommendations: lines.append(f" → {rec}") if self.alternative_models: lines.append("") lines.append("🔄 대안 모델 목록") for model in self.alternative_models: lines.append(f" • {model}") if self.regime_strategy_hint: lines.append("") lines.append(f"📌 레짐 전략: {self.regime_strategy_hint}") lines.append("=" * 55) return "\n".join(lines) def is_critical(self) -> bool: """즉각적인 조치가 필요한 수준인지 (텔레그램 알림 기준)""" if not self.model_suitable and self.suitability_score < 0.4: return True if self.backtest_result and self.backtest_result.sharpe_ratio < 0: return True if self.backtest_result and self.backtest_result.max_drawdown_pct < -30: return True return False class ModelValidator: """ LSTM v3 모델 검증기 검증 흐름: 1. 시장 레짐 감지 (코스피 수준) 2. 백테스트 실행 (선택) 3. 레짐별 모델 적합성 평가 4. 종합 보고서 생성 5. 심각한 경우 텔레그램 알림 """ _CACHE_FILE = "model_validation_cache.json" def __init__(self): self._cache_path = os.path.join(Config.DATA_DIR, self._CACHE_FILE) self._cache: dict = self._load_cache() def _load_cache(self) -> dict: if os.path.exists(self._cache_path): try: with open(self._cache_path, "r", encoding="utf-8") as f: return json.load(f) except Exception: pass return {} def _save_cache(self): try: with open(self._cache_path, "w", encoding="utf-8") as f: json.dump(self._cache, f, ensure_ascii=False, indent=2) except Exception as e: print(f"[Validator] 캐시 저장 실패: {e}") def validate( self, ticker: str, ohlcv_data: dict, strategy_fn=None, kospi_price: float = 2500.0, kospi_change_pct: float = 0.0, run_backtest: bool = True, ) -> ValidationReport: """ 모델 검증 실행 Args: ticker: 종목 코드 ohlcv_data: OHLCV 딕셔너리 strategy_fn: 백테스트용 전략 함수 (None이면 백테스트 생략) kospi_price: 현재 코스피 지수 kospi_change_pct: 코스피 당일 등락률 run_backtest: 백테스트 실행 여부 Returns: ValidationReport """ issues: List[str] = [] recommendations: List[str] = [] # ── 1. 시장 레짐 감지 ──────────────────────────────── regime_analysis: RegimeAnalysis = MarketRegimeDetector.detect( kospi_price, kospi_change_pct ) # ── 2. 백테스트 (선택) ─────────────────────────────── backtest_result: Optional[BacktestResult] = None if run_backtest and strategy_fn is not None: try: backtester = Backtester() backtest_result = backtester.run(ohlcv_data, strategy_fn, ticker) except Exception as e: issues.append(f"백테스트 실행 오류: {e}") # ── 3. 백테스트 결과 기준 위반 체크 ───────────────── bt_sharpe = backtest_result.sharpe_ratio if backtest_result else None bt_winrate = backtest_result.win_rate if backtest_result else None bt_mdd = backtest_result.max_drawdown_pct if backtest_result else None bt_pf = backtest_result.profit_factor if backtest_result else None if backtest_result: if bt_sharpe < _MIN_SHARPE: issues.append( f"Sharpe Ratio 미흡: {bt_sharpe:.3f} (최소 {_MIN_SHARPE})" ) recommendations.append("LSTM 피처 확장 또는 모델 아키텍처 재검토") if bt_winrate < _MIN_WIN_RATE: issues.append( f"승률 미흡: {bt_winrate:.1f}% (최소 {_MIN_WIN_RATE:.0f}%)" ) recommendations.append("매수 진입 임계값 상향 조정 (+0.05)") if bt_mdd < _MAX_MDD: issues.append( f"MDD 과다: {bt_mdd:.2f}% (허용 {_MAX_MDD:.0f}%)" ) recommendations.append("ATR 손절 배수 축소 (ATR×2 → ATR×1.5)") if bt_pf < _MIN_PROFIT_FACTOR: issues.append( f"손익비 미흡: {bt_pf:.2f} (최소 {_MIN_PROFIT_FACTOR})" ) recommendations.append("익절 배수 확대 (ATR×3 → ATR×4)") # ── 4. 레짐 기반 모델 적합성 평가 ─────────────────── regime_validation = MarketRegimeDetector.validate_model_for_regime( regime_analysis.regime, backtest_sharpe=bt_sharpe, backtest_winrate=bt_winrate, backtest_mdd=bt_mdd, ) if not regime_validation["is_suitable"]: issues.append( f"레짐 부적합: {regime_analysis.regime.value} 환경에서 " f"LSTM v3 한계 감지" ) recommendations.append(regime_validation["recommendation"]) # 코스피 6300 특별 경고 if kospi_price >= 5000: issues.append( f"⚠️ 코스피 {kospi_price:.0f} - 역사적 극고점 수준 " "LSTM 비선형 패턴 포착 한계 주의" ) recommendations.append( "Temporal Fusion Transformer(TFT) 또는 Mamba 모델 전환 검토" ) # ── 5. 종합 적합성 점수 ────────────────────────────── suitability_score = regime_validation["confidence_score"] # 문제 건수에 따라 감점 (건당 10%, 최대 50% 감점) penalty = min(len(issues) * 0.10, 0.50) suitability_score = max(0.0, suitability_score - penalty) suitability_score = round(suitability_score, 3) # ── 6. 보고서 생성 ─────────────────────────────────── report = ValidationReport( ticker=ticker, kospi_level=kospi_price, regime=regime_analysis.regime.value, regime_description=regime_analysis.description, backtest_result=backtest_result, model_suitable=(suitability_score >= 0.5 and not regime_validation["should_replace"]), suitability_score=suitability_score, issues=issues, recommendations=list(set(recommendations)), # 중복 제거 alternative_models=regime_validation.get("alternative_models", []), regime_strategy_hint=regime_analysis.recommended_strategy, risk_level=regime_analysis.risk_level, ) # ── 7. 캐시 저장 ───────────────────────────────────── self._cache[ticker] = { "timestamp": time.time(), "kospi_level": kospi_price, "regime": regime_analysis.regime.value, "suitability_score": suitability_score, "should_replace": regime_validation["should_replace"], "issue_count": len(issues), } self._save_cache() return report def get_cached(self, ticker: str) -> Optional[dict]: """캐시된 검증 결과 반환 (24시간 이내)""" cached = self._cache.get(ticker) if not cached: return None if time.time() - cached.get("timestamp", 0) > _CACHE_TTL_SECONDS: return None return cached def send_alert(self, report: ValidationReport): """심각한 검증 결과 텔레그램 알림""" if not report.is_critical(): return try: from modules.services.telegram import TelegramMessenger msg = ( f"🚨 [모델 경고] {report.ticker}\n" f"코스피 {report.kospi_level:.0f} | 레짐: {report.regime}\n" f"적합성: {report.suitability_score:.0%}\n" ) if report.issues: msg += "문제:\n" + "\n".join(f"• {i}" for i in report.issues[:3]) if report.alternative_models: msg += f"\n권고 모델: {report.alternative_models[0]}" TelegramMessenger().send_message(msg) except Exception: pass def generate_regime_report(self, kospi_price: float) -> str: """코스피 수준만으로 빠른 레짐 보고서 생성 (백테스트 없음)""" regime_analysis = MarketRegimeDetector.detect(kospi_price) validation = MarketRegimeDetector.validate_model_for_regime(regime_analysis.regime) lines = [ "=" * 55, f"📈 코스피 {kospi_price:.0f} 레짐 분석", "=" * 55, f"레짐 : {regime_analysis.regime.value}", f"설명 : {regime_analysis.description}", f"리스크 수준 : {regime_analysis.risk_level}", "", "─ 전략 파라미터 조정 ─", f"매수 임계값 : {'+' if regime_analysis.buy_threshold_adj >= 0 else ''}" f"{regime_analysis.buy_threshold_adj:+.2f} 조정", f"포지션 크기 : x{regime_analysis.position_size_adj:.2f}", f"LSTM 가중치 : {'+' if regime_analysis.lstm_weight_adj >= 0 else ''}" f"{regime_analysis.lstm_weight_adj:+.2f}", "", "─ 모델 평가 ─", f"현재 모델 적합: {'✅' if validation['is_suitable'] else '⚠️'} " f"(신뢰도 {validation['confidence_score']:.0%})", f"교체 필요 : {'예' if validation['should_replace'] else '아니오'}", f"권고사항 : {validation['recommendation']}", ] if validation["alternative_models"]: lines.append("") lines.append("대안 모델 목록:") for model in validation["alternative_models"]: lines.append(f" • {model}") lines.append("") lines.append(f"📌 전략: {regime_analysis.recommended_strategy}") lines.append("=" * 55) return "\n".join(lines) # 전역 싱글톤 _validator_instance: Optional[ModelValidator] = None def get_validator() -> ModelValidator: """ModelValidator 싱글톤 반환""" global _validator_instance if _validator_instance is None: _validator_instance = ModelValidator() return _validator_instance