""" Walk-Forward Optimization (WFO) 엔진 롤링 윈도우 기반으로 In-Sample 구간에서 Optuna를 통해 파라미터를 최적화하고, Out-of-Sample 구간에서 검증하여 과적합 여부를 탐지합니다. 핵심 흐름: 1. 전체 기간을 IS(70%) / OOS(30%) 비율로 롤링 윈도우 생성 2. 각 윈도우의 IS 구간에서 Optuna Trial 50회 실행 (목적함수: Sharpe Ratio) 3. 최적 파라미터로 OOS 구간 백테스트 실행 4. IS vs OOS Sharpe 비교 -> 50% 이상 감소 시 과적합 경고 5. 전체 윈도우 종합 결과 반환 의존성: - backtest_engine.py (BacktestEngine, BacktestConfig) - optuna (requirements.txt에 추가 필요) """ import asyncio import logging from dataclasses import dataclass, field, asdict from datetime import date, timedelta from typing import Dict, List, Any, Optional import numpy as np logger = logging.getLogger(__name__) # ── 데이터 클래스 ──────────────────────────────────────────────────────────── @dataclass class WalkForwardConfig: """Walk-Forward 최적화 설정""" stock_codes: List[str] start_date: date end_date: date initial_capital: float = 10_000_000 max_positions: int = 5 position_size: float = 0.2 commission_rate: float = 0.00015 tax_rate: float = 0.0023 # 롤링 윈도우 설정 is_months: int = 6 # In-Sample 기간 (월) oos_months: int = 2 # Out-of-Sample 기간 (월) step_months: int = 1 # 슬라이딩 간격 (월) # Optuna 설정 n_trials: int = 50 # 윈도우당 최대 trial 수 optuna_timeout: int = 300 # 윈도우당 최대 시간 (초) # 과적합 탐지 overfit_threshold: float = 0.5 # IS vs OOS Sharpe 감소율 경고 임계값 @dataclass class WindowResult: """개별 윈도우 결과""" window_index: int is_start: str is_end: str oos_start: str oos_end: str best_params: Dict[str, Any] is_sharpe: float oos_sharpe: float is_return_pct: float oos_return_pct: float is_mdd_pct: float oos_mdd_pct: float is_win_rate: float oos_win_rate: float is_total_trades: int oos_total_trades: int sharpe_degradation: float # (IS - OOS) / |IS|, 높을수록 과적합 overfit_warning: bool # sharpe_degradation > threshold @dataclass class WalkForwardResult: """Walk-Forward 최적화 종합 결과""" config: Dict[str, Any] total_windows: int windows: List[Dict[str, Any]] # 종합 OOS 성과 avg_oos_sharpe: float avg_oos_return_pct: float avg_oos_mdd_pct: float avg_oos_win_rate: float total_oos_trades: int # 과적합 분석 overfit_windows: int # 과적합 경고가 발생한 윈도우 수 avg_sharpe_degradation: float overfit_ratio: float # overfit_windows / total_windows # 최종 추천 파라미터 (OOS Sharpe 가중평균) recommended_params: Dict[str, Any] # ── 유틸리티 ───────────────────────────────────────────────────────────────── def _add_months(d: date, months: int) -> date: """날짜에 월 수를 더합니다. 월말 보정 포함.""" month = d.month - 1 + months year = d.year + month // 12 month = month % 12 + 1 import calendar max_day = calendar.monthrange(year, month)[1] day = min(d.day, max_day) return date(year, month, day) def _generate_windows( start_date: date, end_date: date, is_months: int, oos_months: int, step_months: int, ) -> List[Dict[str, date]]: """롤링 윈도우 목록을 생성합니다. 각 윈도우는 {is_start, is_end, oos_start, oos_end}를 포함합니다. OOS 종료일이 전체 end_date를 초과하면 해당 윈도우부터 중단합니다. """ windows = [] current_start = start_date while True: is_end = _add_months(current_start, is_months) - timedelta(days=1) oos_start = is_end + timedelta(days=1) oos_end = _add_months(oos_start, oos_months) - timedelta(days=1) if oos_end > end_date: # OOS 구간이 전체 기간을 초과하면 end_date로 클램핑 후 마지막 윈도우 if oos_start <= end_date: oos_end = end_date windows.append({ "is_start": current_start, "is_end": is_end, "oos_start": oos_start, "oos_end": oos_end, }) break windows.append({ "is_start": current_start, "is_end": is_end, "oos_start": oos_start, "oos_end": oos_end, }) current_start = _add_months(current_start, step_months) # IS 구간이 전체 기간의 50% 이상을 넘기면 중단 if _add_months(current_start, is_months) > end_date: break return windows # ── Walk-Forward 엔진 ──────────────────────────────────────────────────────── class WalkForwardOptimizer: """Walk-Forward Optimization 엔진 Optuna를 사용하여 각 IS 윈도우에서 파라미터를 최적화하고, OOS 윈도우에서 검증합니다. """ def __init__(self): self._engine_cls = None self._config_cls = None def _lazy_import(self): """BacktestEngine을 실행 시점에 import합니다 (lazy import).""" if self._engine_cls is None: try: from app.services.kr_backtest_engine import ( BacktestEngine as KREngine, MeanReversionBacktest, ) self._engine_cls = KREngine self._config_cls = MeanReversionBacktest logger.info("kr_backtest_engine 로드 성공") except ImportError: from app.services.backtest_engine import ( BacktestEngine, BacktestConfig, ) self._engine_cls = BacktestEngine self._config_cls = BacktestConfig logger.info("kr_backtest_engine 없음 -> backtest_engine 폴백 사용") async def _run_single_backtest( self, stock_codes: List[str], start_date: date, end_date: date, params: Dict[str, Any], wf_config: WalkForwardConfig, ) -> Dict[str, float]: """단일 백테스트를 실행하고 핵심 지표를 반환합니다. Returns: {sharpe_ratio, total_return_pct, mdd_pct, win_rate, total_trades} """ self._lazy_import() from app.services.backtest_engine import BacktestEngine, BacktestConfig config = BacktestConfig( stock_codes=stock_codes, start_date=start_date, end_date=end_date, initial_capital=wf_config.initial_capital, strategy_params=params, max_positions=wf_config.max_positions, position_size=wf_config.position_size, commission_rate=wf_config.commission_rate, tax_rate=wf_config.tax_rate, ) engine = BacktestEngine() result = await engine.run_backtest(config) return { "sharpe_ratio": result.sharpe_ratio, "total_return_pct": result.total_return_pct, "mdd_pct": result.mdd_pct, "win_rate": result.win_rate, "total_trades": result.total_trades, } def _build_params_from_trial(self, trial) -> Dict[str, Any]: """Optuna trial에서 전략 파라미터를 생성합니다. 탐색 범위: - RSI 임계값: 25 ~ 45 - BB 기간: 15 ~ 25 - ATR 배수: 1.5 ~ 3.5 - 손절 %: 4 ~ 10 """ rsi_threshold = trial.suggest_int("rsi_threshold", 25, 45) bb_period = trial.suggest_int("bb_period", 15, 25) atr_multiplier = trial.suggest_float("atr_multiplier", 1.5, 3.5, step=0.1) stop_loss_pct = trial.suggest_float("stop_loss_pct", 4.0, 10.0, step=0.5) return { "MEAN_REVERSION_RSI_THRESHOLD": rsi_threshold, "BB_PERIOD": bb_period, "ATR_STOP_MULTIPLIER": atr_multiplier, "MEAN_REVERSION_STOP_LOSS_PCT": -stop_loss_pct, # 익절은 손절의 2배 (2R 목표) "MEAN_REVERSION_TAKE_PROFIT_PCT": stop_loss_pct * 2.0, "RSI_OVERBOUGHT_LEVEL": 70.0, } async def _optimize_window( self, window_idx: int, window: Dict[str, date], wf_config: WalkForwardConfig, ) -> WindowResult: """단일 윈도우에 대해 IS 최적화 + OOS 검증을 수행합니다.""" try: import optuna optuna.logging.set_verbosity(optuna.logging.WARNING) except ImportError: raise ImportError( "optuna 패키지가 필요합니다. " "'pip install optuna' 를 실행하세요." ) is_start = window["is_start"] is_end = window["is_end"] oos_start = window["oos_start"] oos_end = window["oos_end"] logger.info( f"[윈도우 {window_idx}] IS: {is_start}~{is_end}, " f"OOS: {oos_start}~{oos_end}" ) # ── Optuna 목적함수 (IS 구간 Sharpe Ratio 최대화) ── best_is_metrics = {"sharpe_ratio": -999.0} best_params_holder = [None] async def objective_async(trial) -> float: params = self._build_params_from_trial(trial) try: metrics = await self._run_single_backtest( stock_codes=wf_config.stock_codes, start_date=is_start, end_date=is_end, params=params, wf_config=wf_config, ) except Exception as e: logger.warning(f" Trial {trial.number} 실패: {e}") return -999.0 sharpe = metrics["sharpe_ratio"] # 거래가 너무 적으면 페널티 (최소 3회) if metrics["total_trades"] < 3: sharpe = sharpe - 1.0 # MDD가 -20% 이하이면 페널티 if metrics["mdd_pct"] < -0.20: sharpe = sharpe - 0.5 if sharpe > best_is_metrics["sharpe_ratio"]: best_is_metrics.update(metrics) best_params_holder[0] = params return sharpe # Optuna study 생성 (동기 wrapper로 async objective 실행) study = optuna.create_study( direction="maximize", study_name=f"wfo_window_{window_idx}", sampler=optuna.samplers.TPESampler(seed=42 + window_idx), ) def objective_sync(trial): loop = asyncio.get_event_loop() if loop.is_running(): import concurrent.futures with concurrent.futures.ThreadPoolExecutor() as pool: future = pool.submit( asyncio.run, objective_async(trial) ) return future.result() else: return loop.run_until_complete(objective_async(trial)) # async 환경에서 Optuna 실행 # Optuna는 동기 API이므로 thread에서 실행 def run_study(): study.optimize( objective_sync, n_trials=wf_config.n_trials, timeout=wf_config.optuna_timeout, show_progress_bar=False, ) await asyncio.to_thread(run_study) best_params = best_params_holder[0] if best_params is None: # 모든 trial이 실패한 경우 기본 파라미터 사용 best_params = { "MEAN_REVERSION_RSI_THRESHOLD": 30, "BB_PERIOD": 20, "ATR_STOP_MULTIPLIER": 2.5, "MEAN_REVERSION_STOP_LOSS_PCT": -6.0, "MEAN_REVERSION_TAKE_PROFIT_PCT": 12.0, "RSI_OVERBOUGHT_LEVEL": 70.0, } logger.warning(f"[윈도우 {window_idx}] 모든 trial 실패 -> 기본 파라미터 사용") # ── IS 최종 성과 (best_params) ── is_metrics = await self._run_single_backtest( stock_codes=wf_config.stock_codes, start_date=is_start, end_date=is_end, params=best_params, wf_config=wf_config, ) # ── OOS 검증 ── oos_metrics = await self._run_single_backtest( stock_codes=wf_config.stock_codes, start_date=oos_start, end_date=oos_end, params=best_params, wf_config=wf_config, ) # ── 과적합 탐지 ── is_sharpe = is_metrics["sharpe_ratio"] oos_sharpe = oos_metrics["sharpe_ratio"] if abs(is_sharpe) > 1e-10: sharpe_degradation = (is_sharpe - oos_sharpe) / abs(is_sharpe) else: sharpe_degradation = 0.0 overfit_warning = sharpe_degradation > wf_config.overfit_threshold if overfit_warning: logger.warning( f"[윈도우 {window_idx}] 과적합 경고: " f"IS Sharpe={is_sharpe:.2f}, OOS Sharpe={oos_sharpe:.2f}, " f"감소율={sharpe_degradation:.1%}" ) return WindowResult( window_index=window_idx, is_start=str(is_start), is_end=str(is_end), oos_start=str(oos_start), oos_end=str(oos_end), best_params=best_params, is_sharpe=round(is_sharpe, 4), oos_sharpe=round(oos_sharpe, 4), is_return_pct=round(is_metrics["total_return_pct"], 4), oos_return_pct=round(oos_metrics["total_return_pct"], 4), is_mdd_pct=round(is_metrics["mdd_pct"], 4), oos_mdd_pct=round(oos_metrics["mdd_pct"], 4), is_win_rate=round(is_metrics["win_rate"], 4), oos_win_rate=round(oos_metrics["win_rate"], 4), is_total_trades=is_metrics["total_trades"], oos_total_trades=oos_metrics["total_trades"], sharpe_degradation=round(sharpe_degradation, 4), overfit_warning=overfit_warning, ) def _compute_recommended_params( self, window_results: List[WindowResult] ) -> Dict[str, Any]: """OOS Sharpe 가중평균으로 최종 추천 파라미터를 산출합니다. OOS Sharpe가 양수인 윈도우만 가중치에 포함합니다. 모든 OOS Sharpe가 음수이면 동일 가중치(산술 평균)를 사용합니다. """ param_keys = [ "MEAN_REVERSION_RSI_THRESHOLD", "BB_PERIOD", "ATR_STOP_MULTIPLIER", "MEAN_REVERSION_STOP_LOSS_PCT", "MEAN_REVERSION_TAKE_PROFIT_PCT", "RSI_OVERBOUGHT_LEVEL", ] # 가중치 계산 (OOS Sharpe, 최소 0) weights = [] for wr in window_results: w = max(wr.oos_sharpe, 0.0) weights.append(w) total_weight = sum(weights) # 모든 가중치가 0이면 동일 가중치 if total_weight < 1e-10: weights = [1.0] * len(window_results) total_weight = float(len(window_results)) recommended = {} for key in param_keys: weighted_sum = 0.0 for i, wr in enumerate(window_results): val = wr.best_params.get(key, 0.0) weighted_sum += val * weights[i] raw_val = weighted_sum / total_weight # 정수 파라미터 반올림 if key in ("MEAN_REVERSION_RSI_THRESHOLD", "BB_PERIOD"): recommended[key] = int(round(raw_val)) else: recommended[key] = round(raw_val, 2) return recommended async def run(self, config: WalkForwardConfig) -> WalkForwardResult: """Walk-Forward 최적화를 실행합니다. Args: config: WalkForwardConfig 설정 Returns: WalkForwardResult 종합 결과 """ logger.info( f"Walk-Forward 최적화 시작: {config.start_date} ~ {config.end_date}, " f"IS={config.is_months}개월, OOS={config.oos_months}개월, " f"Step={config.step_months}개월, Trials={config.n_trials}" ) # 1. 롤링 윈도우 생성 windows = _generate_windows( start_date=config.start_date, end_date=config.end_date, is_months=config.is_months, oos_months=config.oos_months, step_months=config.step_months, ) if not windows: raise ValueError( f"유효한 윈도우가 없습니다. 기간이 최소 " f"{config.is_months + config.oos_months}개월 이상이어야 합니다. " f"입력 기간: {config.start_date} ~ {config.end_date}" ) logger.info(f"총 {len(windows)}개 윈도우 생성") # 2. 각 윈도우 순차 실행 (DB 부하 제한) window_results: List[WindowResult] = [] for idx, window in enumerate(windows): try: wr = await self._optimize_window(idx, window, config) window_results.append(wr) logger.info( f"[윈도우 {idx}] 완료 - " f"IS Sharpe={wr.is_sharpe:.2f}, OOS Sharpe={wr.oos_sharpe:.2f}" f"{' [과적합 경고]' if wr.overfit_warning else ''}" ) except Exception as e: logger.error(f"[윈도우 {idx}] 실패: {e}") continue if not window_results: raise RuntimeError("모든 윈도우 최적화가 실패했습니다.") # 3. 종합 OOS 성과 계산 oos_sharpes = [wr.oos_sharpe for wr in window_results] oos_returns = [wr.oos_return_pct for wr in window_results] oos_mdds = [wr.oos_mdd_pct for wr in window_results] oos_winrates = [wr.oos_win_rate for wr in window_results] oos_trades_total = sum(wr.oos_total_trades for wr in window_results) overfit_count = sum(1 for wr in window_results if wr.overfit_warning) degradations = [wr.sharpe_degradation for wr in window_results] # 4. 추천 파라미터 계산 recommended = self._compute_recommended_params(window_results) result = WalkForwardResult( config={ "stock_codes": config.stock_codes, "start_date": str(config.start_date), "end_date": str(config.end_date), "is_months": config.is_months, "oos_months": config.oos_months, "step_months": config.step_months, "n_trials": config.n_trials, "initial_capital": config.initial_capital, }, total_windows=len(window_results), windows=[asdict(wr) for wr in window_results], avg_oos_sharpe=round(float(np.mean(oos_sharpes)), 4), avg_oos_return_pct=round(float(np.mean(oos_returns)), 4), avg_oos_mdd_pct=round(float(np.mean(oos_mdds)), 4), avg_oos_win_rate=round(float(np.mean(oos_winrates)), 4), total_oos_trades=oos_trades_total, overfit_windows=overfit_count, avg_sharpe_degradation=round(float(np.mean(degradations)), 4), overfit_ratio=round(overfit_count / len(window_results), 4), recommended_params=recommended, ) logger.info( f"Walk-Forward 최적화 완료: " f"{len(window_results)}개 윈도우, " f"평균 OOS Sharpe={result.avg_oos_sharpe:.2f}, " f"과적합 비율={result.overfit_ratio:.0%}" ) return result