""" 전략 자동 최적화 서비스 [역할] - 최근 거래 성과 약점 분석 → 파라미터 범위 결정 - BacktestEngine을 사용한 그리드 서치로 최적 파라미터 탐색 - 평가 지표: Sharpe * 0.4 + win_rate * 0.3 + (1 - |MDD|/100) * 0.3 - Sharpe 10% 이상 개선 시 strategy_config 자동 적용 - StrategyOptimizationLog 테이블에 이력 저장 [사용법] from app.services.strategy_optimizer import run_optimization async with AsyncSessionLocal() as db: result = await run_optimization(db, perf_data, trigger="manual") """ import asyncio import logging from datetime import date, datetime, timedelta from typing import Any, Dict, List, Optional, Tuple from dataclasses import dataclass from sqlalchemy import select, desc from sqlalchemy.ext.asyncio import AsyncSession from app.core.database import AsyncSessionLocal from app.models.models import StrategyOptimizationLog, StockDailyPrice from app.services.strategy_config_service import strategy_cfg from app.services.backtest_engine import BacktestEngine, BacktestConfig from app.core.timeutil import now_kst logger = logging.getLogger(__name__) # ───────────────────────────────────────────────────────────────────────────── # 결과 데이터클래스 # ───────────────────────────────────────────────────────────────────────────── @dataclass class OptimizationResult: best_params: Dict[str, Any] best_score: float best_sharpe: float best_win_rate: float best_mdd: float current_sharpe: float previous_params: Dict[str, Any] applied: bool apply_reason: str total_combinations: int # IS/OOS 검증 결과 (v2 추가) oos_sharpe: float = 0.0 oos_win_rate: float = 0.0 oos_mdd: float = 0.0 is_oos_sharpe_ratio: float = 0.0 # OOS Sharpe / IS Sharpe (1.0에 가까울수록 건전) # ───────────────────────────────────────────────────────────────────────────── # 파라미터 범위 결정 (성과 약점 기반) # ───────────────────────────────────────────────────────────────────────────── def _build_param_grid(perf_data: Dict[str, Any]) -> Dict[str, List[Any]]: """ 성과 약점을 분석하여 그리드 서치할 파라미터 범위를 결정. 규칙: - 승률 < 50% → RSI_OVERSOLD_LEVEL 강화 (더 낮은 값) - avg_loss > 손절 기준 → ATR_STOP_MULTIPLIER, STOP_LOSS_PERCENT 조정 - MDD 과다(< -15%) → MAX_POSITION_SIZE, MAX_HOLDINGS_COUNT 조정 - 항상 RSI 기간 변화도 포함 (성과 개선 여지 높음) """ summary = perf_data.get("summary", {}) win_rate = summary.get("win_rate", 50.0) avg_loss = summary.get("avg_loss_pct", -5.0) grid: Dict[str, List[Any]] = {} # RSI_OVERSOLD_LEVEL: 승률 기반 if win_rate < 40: grid["RSI_OVERSOLD_LEVEL"] = [20.0, 25.0, 28.0] elif win_rate < 50: grid["RSI_OVERSOLD_LEVEL"] = [25.0, 28.0, 30.0] else: grid["RSI_OVERSOLD_LEVEL"] = [28.0, 30.0, 32.0] # STOP_LOSS_PERCENT: 평균 손실 기반 if avg_loss < -8.0: grid["STOP_LOSS_PERCENT"] = [-5.0, -6.0, -7.0] elif avg_loss < -6.0: grid["STOP_LOSS_PERCENT"] = [-5.0, -6.0] else: grid["STOP_LOSS_PERCENT"] = [-6.0, -7.0, -8.0] # ATR_STOP_MULTIPLIER: 항상 포함 grid["ATR_STOP_MULTIPLIER"] = [2.0, 2.5, 3.0] # BOLLINGER_STDDEV grid["BOLLINGER_STDDEV"] = [1.8, 2.0, 2.2] return grid def _cartesian_product(grid: Dict[str, List[Any]]) -> List[Dict[str, Any]]: """딕셔너리 그리드의 카르테시안 곱을 반환.""" keys = list(grid.keys()) values = list(grid.values()) def _product(lists): result = [[]] for lst in lists: result = [x + [y] for x in result for y in lst] return result combos = _product(values) return [dict(zip(keys, combo)) for combo in combos] # ───────────────────────────────────────────────────────────────────────────── # 현재 전략 파라미터 수집 # ───────────────────────────────────────────────────────────────────────────── async def _get_current_params() -> Dict[str, Any]: """현재 strategy_cfg 값들을 읽어 반환.""" await strategy_cfg._refresh() return { "RSI_OVERSOLD_LEVEL": strategy_cfg.get("RSI_OVERSOLD_LEVEL", 30.0), "STOP_LOSS_PERCENT": strategy_cfg.get("STOP_LOSS_PERCENT", -6.0), "ATR_STOP_MULTIPLIER": strategy_cfg.get("ATR_STOP_MULTIPLIER", 2.5), "BOLLINGER_STDDEV": strategy_cfg.get("BOLLINGER_STDDEV", 2.0), "MAX_POSITION_SIZE": strategy_cfg.get("MAX_POSITION_SIZE", 0.04), "MAX_HOLDINGS_COUNT": strategy_cfg.get("MAX_HOLDINGS_COUNT", 20), } # ───────────────────────────────────────────────────────────────────────────── # 백테스트로 파라미터 평가 # ───────────────────────────────────────────────────────────────────────────── def _compute_score(sharpe: float, win_rate: float, mdd: float) -> float: """통합 평가 지표 계산: Sharpe*0.4 + win_rate*0.3 + (1-|mdd|/100)*0.3""" mdd_penalty = max(0.0, 1.0 - abs(mdd) / 100.0) return sharpe * 0.4 + (win_rate / 100.0) * 0.3 + mdd_penalty * 0.3 async def _evaluate_params( params: Dict[str, Any], stock_codes: List[str], start_date: date, end_date: date, initial_capital: float, ) -> Tuple[float, float, float, float]: """ 단일 파라미터 조합으로 백테스트 실행. Returns: (score, sharpe, win_rate, mdd) """ try: backtest_params = { "MEAN_REVERSION_RSI_THRESHOLD": params.get("RSI_OVERSOLD_LEVEL", 30.0), "MEAN_REVERSION_STOP_LOSS_PCT": params.get("STOP_LOSS_PERCENT", -6.0), "MAX_TOTAL_POSITIONS": min(5, len(stock_codes)), } config = BacktestConfig( stock_codes=stock_codes, start_date=start_date, end_date=end_date, initial_capital=initial_capital, strategy_params=backtest_params, max_positions=min(5, len(stock_codes)), position_size=params.get("MAX_POSITION_SIZE", 0.2), ) engine = BacktestEngine() result = await engine.run_backtest(config) sharpe = float(result.sharpe_ratio or 0.0) win_rate = float(result.win_rate or 0.0) mdd = float(result.mdd_pct or 0.0) score = _compute_score(sharpe, win_rate, mdd) return score, sharpe, win_rate, mdd except Exception as e: logger.debug(f"백테스트 실패 (파라미터: {params}): {e}") return -1.0, 0.0, 0.0, 0.0 # ───────────────────────────────────────────────────────────────────────────── # 대표 종목 선택 # ───────────────────────────────────────────────────────────────────────────── async def _get_representative_stocks(period_days: int = 90) -> List[str]: """ 최근 N일 거래가 있는 대표 종목 10개 반환. 백테스트 실행 시간을 줄이기 위해 종목 수 제한. """ from app.models.models import TradeHistory from sqlalchemy import func, and_ from datetime import datetime, timedelta since = now_kst() - timedelta(days=period_days) async with AsyncSessionLocal() as db: stmt = ( select(TradeHistory.stock_code, func.count(TradeHistory.id).label("cnt")) .where(TradeHistory.created_at >= since) .group_by(TradeHistory.stock_code) .order_by(desc("cnt")) .limit(10) ) result = await db.execute(stmt) codes = [row[0] for row in result.fetchall()] # 거래 이력 없으면 stock_daily_prices에서 최근 종목 fallback if not codes: async with AsyncSessionLocal() as db: stmt = ( select(StockDailyPrice.stock_code) .where(StockDailyPrice.trade_date >= (date.today() - timedelta(days=period_days))) .group_by(StockDailyPrice.stock_code) .limit(10) ) result = await db.execute(stmt) codes = [row[0] for row in result.fetchall()] return codes or [] # ───────────────────────────────────────────────────────────────────────────── # 파라미터 적용 # ───────────────────────────────────────────────────────────────────────────── async def apply_params(params: Dict[str, Any], reason: str = "auto_optimizer") -> bool: """strategy_cfg에 파라미터를 일괄 적용.""" try: for key, value in params.items(): await strategy_cfg.upsert(key, str(value), updated_by=reason) logger.info(f"✅ 전략 파라미터 자동 적용 완료: {list(params.keys())} (사유: {reason})") return True except Exception as e: logger.error(f"❌ 파라미터 적용 실패: {e}", exc_info=True) return False # ───────────────────────────────────────────────────────────────────────────── # 최적화 이력 저장 # ───────────────────────────────────────────────────────────────────────────── async def save_optimization_log(db: AsyncSession, result: OptimizationResult, trigger: str) -> StrategyOptimizationLog: """최적화 결과를 DB에 저장.""" log = StrategyOptimizationLog( run_date=date.today(), trigger=trigger, period_days=90, best_params=result.best_params, best_sharpe=result.best_sharpe, best_win_rate=result.best_win_rate, best_mdd=result.best_mdd, current_sharpe=result.current_sharpe, previous_params=result.previous_params, applied=result.applied, apply_reason=result.apply_reason, total_combinations=result.total_combinations, ) db.add(log) await db.commit() await db.refresh(log) return log # ───────────────────────────────────────────────────────────────────────────── # 메인 최적화 함수 # ───────────────────────────────────────────────────────────────────────────── async def run_optimization( db: AsyncSession, perf_data: Dict[str, Any], trigger: str = "auto", period_days: int = 90, initial_capital: float = 10_000_000, ) -> OptimizationResult: """ 성과 데이터 기반 전략 파라미터 최적화. Args: db: AsyncSession perf_data: analyze_performance() 반환값 trigger: "auto" | "manual" period_days: 백테스트 기간 (일) initial_capital: 백테스트 초기 자본 Returns: OptimizationResult """ logger.info(f"전략 최적화 시작 (trigger={trigger}, period={period_days}일)") # 1. 현재 파라미터 저장 (롤백용) prev_params = await _get_current_params() # 2. 그리드 구성 param_grid = _build_param_grid(perf_data) combinations = _cartesian_product(param_grid) total_combos = len(combinations) logger.info(f" 그리드 조합 수: {total_combos}") # 3. 대표 종목 선택 stock_codes = await _get_representative_stocks(period_days) if not stock_codes: logger.warning("백테스트 대상 종목 없음. 최적화 건너뜀.") return OptimizationResult( best_params=prev_params, best_score=0.0, best_sharpe=0.0, best_win_rate=0.0, best_mdd=0.0, current_sharpe=0.0, previous_params=prev_params, applied=False, apply_reason="no_stock_data", total_combinations=0, ) # ── IS/OOS 기간 분할 (70% IS / 30% OOS, 날짜 기준) ── end_date = date.today() start_date = end_date - timedelta(days=period_days) is_days = int(period_days * 0.7) is_end_date = start_date + timedelta(days=is_days) oos_start_date = is_end_date + timedelta(days=1) oos_end_date = end_date logger.info( f" IS 기간: {start_date} ~ {is_end_date} ({is_days}일), " f"OOS 기간: {oos_start_date} ~ {oos_end_date} ({period_days - is_days}일)" ) # 4. 현재 파라미터 기준 Sharpe 계산 (IS 기간) cur_score, cur_sharpe, cur_wr, cur_mdd = await _evaluate_params( prev_params, stock_codes, start_date, is_end_date, initial_capital ) logger.info(f" 현재 IS 성과 -> Sharpe: {cur_sharpe:.3f}, 승률: {cur_wr:.1f}%, MDD: {cur_mdd:.1f}%") # 5. 그리드 서치 — IS 기간에서 최적 파라미터 탐색 (배치 5개씩 병렬) best_score = cur_score best_params = dict(prev_params) best_sharpe = cur_sharpe best_win_rate = cur_wr best_mdd = cur_mdd batch_size = 5 for i in range(0, len(combinations), batch_size): batch = combinations[i:i + batch_size] tasks = [ _evaluate_params(p, stock_codes, start_date, is_end_date, initial_capital) for p in batch ] results = await asyncio.gather(*tasks, return_exceptions=True) for combo, res in zip(batch, results): if isinstance(res, Exception): continue score, sharpe, wr, mdd = res if score > best_score: best_score = score best_sharpe = sharpe best_win_rate = wr best_mdd = mdd best_params = {**prev_params, **combo} logger.info(f" IS 최적 성과 -> Sharpe: {best_sharpe:.3f}, 승률: {best_win_rate:.1f}%, MDD: {best_mdd:.1f}%") # 6. OOS 기간에서 최적 파라미터 검증 (과적합 탐지) oos_score, oos_sharpe, oos_wr, oos_mdd = await _evaluate_params( best_params, stock_codes, oos_start_date, oos_end_date, initial_capital ) is_oos_ratio = (oos_sharpe / best_sharpe) if best_sharpe > 0 else 0.0 logger.info( f" OOS 검증 -> Sharpe: {oos_sharpe:.3f}, 승률: {oos_wr:.1f}%, MDD: {oos_mdd:.1f}%, " f"IS/OOS Sharpe 비율: {is_oos_ratio:.2f}" ) if is_oos_ratio < 0.5 and best_sharpe > 0: logger.warning( f" [과적합 경고] OOS Sharpe / IS Sharpe = {is_oos_ratio:.2f} < 0.5 " "-- 파라미터가 IS 기간에 과적합되었을 가능성 높음" ) # 7. 자동 적용 여부 결정 # 조건: IS Sharpe 10% 이상 개선 AND OOS Sharpe > 0 (OOS에서도 양의 성과) improvement_threshold = 1.1 _is_improved = ( cur_sharpe > 0 and best_sharpe >= cur_sharpe * improvement_threshold and best_params != prev_params ) if cur_sharpe <= 0 and best_sharpe > 0: _is_improved = True # OOS 검증 게이트: OOS Sharpe > 0 이어야 실전 적용 허가 _oos_passed = oos_sharpe > 0 _should_apply = _is_improved and _oos_passed applied = False apply_reason = "" if _should_apply: improvement_pct = ((best_sharpe - cur_sharpe) / max(abs(cur_sharpe), 0.001)) * 100 apply_reason = ( f"IS Sharpe {improvement_pct:.1f}% 향상 ({cur_sharpe:.3f} -> {best_sharpe:.3f}), " f"OOS Sharpe: {oos_sharpe:.3f}, IS/OOS 비율: {is_oos_ratio:.2f}" ) # 최적화된 파라미터만 적용 (prev와 다른 키만) changed = {k: v for k, v in best_params.items() if str(v) != str(prev_params.get(k, ""))} if changed: applied = await apply_params(changed, reason="auto_optimizer") if applied: logger.info(f"자동 적용 완료: {apply_reason}") else: if cur_sharpe <= 0 and best_sharpe <= 0: apply_reason = "insufficient_data" elif not _oos_passed and _is_improved: apply_reason = f"OOS 검증 실패 (OOS Sharpe: {oos_sharpe:.3f} <= 0, 과적합 의심)" elif best_sharpe < cur_sharpe * improvement_threshold: pct = ((best_sharpe - cur_sharpe) / max(abs(cur_sharpe), 0.001)) * 100 apply_reason = f"IS 개선 미달 ({pct:+.1f}% < 10%)" else: apply_reason = "no_improvement" logger.info(f"자동 적용 건너뜀: {apply_reason}") return OptimizationResult( best_params=best_params, best_score=best_score, best_sharpe=best_sharpe, best_win_rate=best_win_rate, best_mdd=best_mdd, current_sharpe=cur_sharpe, previous_params=prev_params, applied=applied, apply_reason=apply_reason, total_combinations=total_combos, oos_sharpe=oos_sharpe, oos_win_rate=oos_wr, oos_mdd=oos_mdd, is_oos_sharpe_ratio=round(is_oos_ratio, 4), ) # ───────────────────────────────────────────────────────────────────────────── # 마지막 최적화 이력 조회 # ───────────────────────────────────────────────────────────────────────────── async def get_last_optimization_date(db: AsyncSession) -> Optional[date]: """마지막 최적화 실행일 반환 (없으면 None).""" stmt = select(StrategyOptimizationLog.run_date).order_by( desc(StrategyOptimizationLog.created_at) ).limit(1) result = await db.execute(stmt) row = result.scalar_one_or_none() return row