gahusb
7ea1a21487
Atomic mv of root V1 assets (main_server.py + modules/ + data/ + tests/ + entry scripts + docs + logs) into signal_v1/ subdirectory. load_dotenv() updated to load web-ai/.env explicitly via Path. Adds web-ai/CLAUDE.md (workspace guide) and web-ai/start.bat (signal_v1 entry wrapper). Prepares for signal_v2/ Phase 2. Tests: signal_v1/tests/unit baseline preserved (no regression). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
349 lines
14 KiB
Python
349 lines
14 KiB
Python
"""
|
||
모델 검증 시스템 (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
|