""" 그리드 서치를 통한 모델 최적화 모듈 """ import numpy as np import itertools import tensorflow as tf from pathlib import Path import matplotlib.pyplot as plt from ..models.contime import build_contime_lstm_model from ..data.processors import prepare_data from ..data.normalize import clean_numeric_data from ..evaluation.backtest import backtest_by_ticker, get_risk_free_rate from ..evaluation.model_evaluation import evaluate_model, calculate_combined_score from ..data.hierarchical_embedding import create_sector_industry_mapping, apply_sector_industry_mapping from .utils import TqdmProgressCallback, save_model, save_results, save_metadata, get_project_root from ..visualization.plots import plot_training_history, plot_performance_grid, plot_signal_distribution, plot_price_predictions, plot_graph_embeddings def evaluate_config(config, data_dict, ticker_encoder, risk_free_rate, sector_industry_df=None, selection_method='combined_score'): """ 특정 설정에 대한 모델을 훈련하고 평가합니다. """ try: # 데이터 추출 x_train = data_dict['x_train'] y_train = data_dict['y_train'] ticker_train = data_dict['ticker_train'] y_train_dt = data_dict['y_train_dt'] x_val = data_dict['x_val'] y_val = data_dict['y_val'] ticker_val = data_dict['ticker_val'] y_val_dt = data_dict['y_val_dt'] # 특성 데이터 정리 x_train = clean_numeric_data(x_train, verbose=False) x_val = clean_numeric_data(x_val, verbose=False) # 섹터/산업 정보 추출 sector_train = np.zeros_like(ticker_train) industry_train = np.zeros_like(ticker_train) sector_val = np.zeros_like(ticker_val) industry_val = np.zeros_like(ticker_val) if sector_industry_df is not None: id_to_ticker = {v: k for k, v in ticker_encoder.mapping.items()} ticker_list = list(id_to_ticker.values()) sector_mapping, industry_mapping = create_sector_industry_mapping( ticker_list, sector_industry_df ) sector_train, industry_train = apply_sector_industry_mapping( ticker_train, ticker_encoder, sector_mapping, industry_mapping ) sector_val, industry_val = apply_sector_industry_mapping( ticker_val, ticker_encoder, sector_mapping, industry_mapping ) # 기존 섹터/산업 데이터 사용 if 'sector_train' in data_dict and 'industry_train' in data_dict: sector_train = data_dict['sector_train'] industry_train = data_dict['industry_train'] sector_val = data_dict['sector_val'] industry_val = data_dict['industry_val'] # 데이터 타입 변환 inputs_train = [ np.asarray(x_train, dtype=np.float32), np.asarray(ticker_train, dtype=np.int32), np.asarray(sector_train, dtype=np.int32), np.asarray(industry_train, dtype=np.int32), np.asarray(data_dict['time_diffs_train'], dtype=np.float32) ] inputs_val = [ np.asarray(x_val, dtype=np.float32), np.asarray(ticker_val, dtype=np.int32), np.asarray(sector_val, dtype=np.int32), np.asarray(industry_val, dtype=np.int32), np.asarray(data_dict['time_diffs_val'], dtype=np.float32) ] targets_train = { 'value_output': np.asarray(y_train, dtype=np.float32), 'derivative_output': np.asarray(y_train_dt, dtype=np.float32) } targets_val = { 'value_output': np.asarray(y_val, dtype=np.float32), 'derivative_output': np.asarray(y_val_dt, dtype=np.float32) } # 모델 생성 model = build_contime_lstm_model( seq_len=x_train.shape[1], num_features=x_train.shape[2], hidden_dim=config['hidden_dim'], dropout_rate=config['dropout_rate'], num_tickers=len(np.unique(ticker_train)), dt=config['dt'], ode_steps=config['ode_steps'], value_weight=config['value_weight'], derivative_weight=config['derivative_weight'], num_sectors=len(np.unique(sector_train)), num_industries=len(np.unique(industry_train)), ) # 콜백 정의 callbacks = [ tf.keras.callbacks.EarlyStopping( monitor='val_loss', patience=config['patience'], restore_best_weights=True, verbose=0 ), tf.keras.callbacks.ReduceLROnPlateau( monitor='val_loss', factor=config['factor'], patience=config['patience'] // 2, min_lr=config['min_lr'], verbose=0 ), TqdmProgressCallback(epochs=config['epochs']) ] # 학습 history = model.fit( inputs_train, targets_train, validation_data=(inputs_val, targets_val), epochs=config['epochs'], batch_size=config['batch_size'], callbacks=callbacks, verbose=2 ) # 평가 metrics = evaluate_model( model, x_val, y_val, ticker_val, y_val_dt, sector_test=sector_val, industry_test=industry_val, time_diffs_test=inputs_val[4], verbose=False ) # 예측 수행 pred_val = model.predict(inputs_val, verbose=0) # 예측값 처리 y_pred_val = pred_val[0] if isinstance(pred_val, list) else pred_val # 마지막 타임스텝 추출 if len(y_pred_val.shape) == 3: y_pred_val = y_pred_val[:, -1, 0] else: y_pred_val = y_pred_val.flatten() # 데이터 정리 y_pred_val = np.asarray(y_pred_val).flatten() y_val_flat = np.asarray(y_val).flatten() ticker_val_flat = np.asarray(ticker_val).flatten() # 길이 맞추기 min_len = min(len(y_pred_val), len(y_val_flat), len(ticker_val_flat)) y_pred_val = y_pred_val[:min_len] y_val_flat = y_val_flat[:min_len] ticker_val_flat = ticker_val_flat[:min_len] # 거래 기회 계산 num_tickers = len(np.unique(ticker_val_flat)) trading_days = len(y_val_flat) // num_tickers total_opportunities = trading_days * num_tickers min_expected_trades = max(10, int(total_opportunities * 0.05)) # 최적 임계값 찾기 use_combined_score = (selection_method == 'combined_score') best_threshold, best_backtest, all_thresholds = find_optimal_threshold( y_pred_val, y_val_flat, ticker_val_flat, risk_free_rate, min_expected_trades, use_combined_score ) # 메트릭 계산 if selection_method == 'combined_score': combined_metric = calculate_combined_score( best_backtest, min_trades=50, max_trades=150 ) metrics.update({ 'combined_score': combined_metric, 'best_threshold': best_threshold, 'total_return': best_backtest['portfolio']['total_return'], 'sharpe_ratio': best_backtest['portfolio']['sharpe_ratio'], 'max_drawdown': best_backtest['portfolio']['max_drawdown'], 'trade_count': len(best_backtest['portfolio'].get('trades', [])), 'win_rate': best_backtest['portfolio'].get('win_rate', 0), 'avg_ticker_sharpe': best_backtest['avg_ticker_sharpe'] }) else: # 기존 방식 유지 metrics.update({ 'best_threshold': best_threshold, 'total_return': best_backtest['portfolio']['total_return'], 'sharpe_ratio': best_backtest['portfolio']['sharpe_ratio'], 'max_drawdown': best_backtest['portfolio']['max_drawdown'], 'trade_count': len(best_backtest['portfolio'].get('trades', [])), 'win_rate': best_backtest['portfolio'].get('win_rate', 0), 'avg_ticker_sharpe': best_backtest['avg_ticker_sharpe'] }) print(f"임계값: {best_threshold:.4f}, 수익률: {best_backtest['portfolio']['total_return']:.4f}") print(f"거래: {len(best_backtest['portfolio'].get('trades', []))}/{total_opportunities} " f"({len(best_backtest['portfolio'].get('trades', [])) / total_opportunities:.1%})") if selection_method == 'combined_score': print(f"복합 점수: {combined_metric:.4f}") return { 'config': config, 'metrics': metrics, 'model': model, 'history': history.history, 'best_threshold': best_threshold, 'ticker_metrics': best_backtest['by_ticker'], 'total_opportunities': total_opportunities, 'min_expected_trades': min_expected_trades, 'all_thresholds': all_thresholds } except Exception as e: print(f"모델 평가 실패: {str(e)}") return None def find_optimal_threshold(y_pred_val, y_val_flat, ticker_val_flat, risk_free_rate, min_expected_trades, use_combined_score=True): """ 최적 임계값을 찾는 헬퍼 함수 """ thresholds = np.arange(0.00, 0.05, 0.001) best_weighted_score = -np.inf best_threshold = 0 best_backtest = None # 모든 임계값 결과 저장 all_thresholds = {} for threshold in thresholds: result = backtest_by_ticker( predictions=y_pred_val, actual_returns=y_val_flat, ticker_ids=ticker_val_flat, threshold=threshold, commission=0.0025, risk_free_rate=risk_free_rate ) # 모든 임계값 결과 저장 all_thresholds[float(threshold)] = { 'total_return': result['portfolio']['total_return'], 'sharpe_ratio': result['portfolio']['sharpe_ratio'], 'max_drawdown': result['portfolio']['max_drawdown'], 'trades': result['portfolio'].get('trades', []) } trade_count = len(result['portfolio'].get('trades', [])) if use_combined_score: min_trades = max(10, min_expected_trades // 2) # 최소 거래수 조정 max_trades = min_expected_trades * 2 # 최대 거래수 설정 weighted_score = calculate_combined_score( result, min_trades=min_trades, max_trades=max_trades ) else: trade_ratio_score = min(1.0, trade_count / min_expected_trades) if trade_count >= (min_expected_trades * 0.5) else (trade_count / min_expected_trades) ** 2 weighted_score = result['avg_ticker_sharpe'] * trade_ratio_score if weighted_score > best_weighted_score: best_weighted_score = weighted_score best_threshold = threshold best_backtest = result return best_threshold, best_backtest, all_thresholds def run_optimization_pipeline(data_dict, ticker_encoder, metric='combined_score', output_path=None, save=True, model_output=None, sector_industry_df=None, run_visualizations=False): """ 연속 시간 모델 최적화 파이프라인 """ print("===== 연속 시간 모델 최적화 =====") print(f"선택 기준: {metric}") # 무위험 수익률 계산 start_date = data_dict.get('start_date') end_date = data_dict.get('end_date') risk_free_rate = get_risk_free_rate(start_date, end_date) print(f"무위험 수익률: {risk_free_rate:.6f}") # 시각화 저장 경로 설정 plots_dir = Path(get_project_root()) / "models" / "plots" if run_visualizations: plots_dir.mkdir(parents=True, exist_ok=True) print(f"시각화 결과는 {plots_dir}에 저장됩니다.") # 파라미터 그리드 param_grid = { 'hidden_dim': [128], 'dropout_rate': [0.3], 'dt': [0.1], 'ode_steps': [5], 'value_weight': [0.8], 'factor': [0.5], 'patience': [10], 'min_lr': [1e-6], 'epochs': [1], 'batch_size': [64] } # param_grid = { # 'hidden_dim': [96, 128, 256], # 'dropout_rate': [0.3], # 'dt': [0.1], # 'ode_steps': [5], # 'value_weight': [0.5, 0.6, 0.7, 0.8, 0.9], # 'factor': [0.5], # 'patience': [10], # 'min_lr': [1e-6], # 'epochs': [50], # 'batch_size': [64, 96] # } # 데이터 유효성 확인 required_keys = ['x_train', 'y_train', 'ticker_train', 'time_diffs_train', 'x_val', 'y_val', 'ticker_val', 'time_diffs_val'] if not all(key in data_dict for key in required_keys): print("필요한 데이터 키가 없습니다. 데이터를 준비합니다...") data_dict, _, _ = prepare_data(data_dict.get('data'), window_size=60) # 그리드 서치 실행 param_keys = list(param_grid.keys()) param_values = list(param_grid.values()) total_combinations = 1 for values in param_values: total_combinations *= len(values) print(f"그리드 서치 실행: 총 {total_combinations}개의 파라미터 조합을 테스트합니다.") # 결과 저장 results = [] best_score = -float('inf') best_config = None iteration_counter = 0 # 모든 조합 생성 및 테스트 for combination_values in itertools.product(*param_values): config = dict(zip(param_keys, combination_values)) config['derivative_weight'] = 1.0 - config['value_weight'] iteration_counter += 1 print("\n" + "=" * 60) print(f"조합 {iteration_counter}/{total_combinations}") print(f"현재 파라미터:") for k, v in config.items(): print(f" {k}: {v}") result = evaluate_config( config, data_dict, ticker_encoder, risk_free_rate, sector_industry_df, selection_method=metric ) if result is None: print(f"설정 {iteration_counter}에 대한 평가 결과가 None입니다.") continue results.append(result) if metric == 'combined_score' and 'combined_score' in result['metrics']: metric_value = result['metrics']['combined_score'] else: metric_value = result['metrics'].get(metric, 0) current_return = result['metrics'].get('total_return', 0) current_sharpe = result['metrics'].get('sharpe_ratio', 0) current_trades = result['metrics'].get('trade_count', 0) # 복합 점수도 출력 if 'combined_score' in result['metrics']: combined_score = result['metrics']['combined_score'] print(f"결과 - {metric}: {metric_value:.4f}, 복합점수: {combined_score:.4f}, " f"수익률: {current_return:.4f}, 샤프: {current_sharpe:.4f}, 거래: {current_trades}") else: print(f"결과 - {metric}: {metric_value:.4f}, 수익률: {current_return:.4f}, " f"샤프: {current_sharpe:.4f}, 거래: {current_trades}") if metric_value > best_score: best_score = metric_value best_config = config print(f"새로운 최고 성능 발견! {metric}: {best_score:.4f}") print(f" 수익률: {current_return:.4f}") print(f" 샤프 비율: {current_sharpe:.4f}") print(f" 거래 수: {current_trades}") else: print(f"현재 최고 성능 ({metric}): {best_score:.4f}") if not results: print("모든 설정에서 평가가 실패했습니다.") return {'error': '모든 모델 평가 실패', 'best_config': None, 'results': []} best_result = max(results, key=lambda x: x['metrics'].get(metric, 0)) best_config = best_result['config'] best_threshold = best_result['best_threshold'] best_model = best_result['model'] print(f"\n그리드 서치 완료!") print(f" 총 {len(results)}개 결과 중 최고 성능:") print(f" {metric}: {best_result['metrics'].get(metric, 0):.4f}") print(f" 수익률: {best_result['metrics'].get('total_return', 0):.4f}") print(f" 샤프 비율: {best_result['metrics'].get('sharpe_ratio', 0):.4f}") print(f" 거래 수: {best_result['metrics'].get('trade_count', 0)}") # 종목별 메트릭 출력 ticker_metrics = best_result['ticker_metrics'] ticker_ids = list(ticker_metrics.keys()) n_tickers = len(ticker_ids) # 종목별 메트릭 추출 및 평균 계산 avg_return = np.mean([ticker_metrics[tid]['total_return'] for tid in ticker_ids]) avg_sharpe = np.mean([ticker_metrics[tid]['sharpe_ratio'] for tid in ticker_ids]) avg_mdd = np.mean([ticker_metrics[tid]['max_drawdown'] for tid in ticker_ids]) avg_win_rate = np.mean([ticker_metrics[tid].get('win_rate', 0) for tid in ticker_ids]) # 거래 수 계산 total_trades = sum([len(ticker_metrics[tid].get('trades', [])) for tid in ticker_ids]) avg_trades = total_trades / n_tickers print("\n===== 최적 설정 =====") print(best_config) print(f"최적 임계값: {best_threshold:.4f}") print(f"\n----- 종목별 평균 성능 (티커 수: {n_tickers}) -----") print(f"평균 종목 수익률: {avg_return:.4f}") print(f"평균 종목 샤프 비율: {avg_sharpe:.4f}") print(f"평균 종목 최대 낙폭: {avg_mdd:.4f}") print(f"평균 종목 승률: {avg_win_rate:.2%}") print(f"평균 종목 거래 횟수: {avg_trades:.1f}\n") # 시각화 if run_visualizations: try: # 1. 학습 기록 시각화 fig1 = plot_training_history(best_result['history']) if fig1: fig1.savefig(plots_dir / "training_history.png", dpi=300, bbox_inches='tight') plt.close(fig1) print(" - training_history.png 저장 완료") # 2. 성능 그리드 시각화 fig2 = plot_performance_grid({0.0025: best_result.get('all_thresholds', {})}) if fig2: fig2.savefig(plots_dir / "performance_grid.png", dpi=300, bbox_inches='tight') plt.close(fig2) print(" - performance_grid.png 저장 완료") # 예측 및 시각화를 위한 데이터 준비 x_val_clean = clean_numeric_data(data_dict['x_val'], replace_nan=0.0, replace_inf=0.0, verbose=False) ticker_val = np.asarray(data_dict['ticker_val'], dtype=np.int32) # 섹터/산업 데이터와 시간 간격 데이터 처리 sector_val = data_dict.get('sector_val', np.zeros_like(ticker_val)) industry_val = data_dict.get('industry_val', np.zeros_like(ticker_val)) time_diffs_val = np.asarray(data_dict['time_diffs_val'], dtype=np.float32) # 예측 수행 pred_val = best_model.predict([ tf.cast(x_val_clean, tf.float32), tf.cast(ticker_val, tf.int32), tf.cast(sector_val, tf.int32), tf.cast(industry_val, tf.int32), tf.cast(time_diffs_val, tf.float32) ], verbose=0) if isinstance(pred_val, list): y_pred_val = pred_val[0].flatten() else: y_pred_val = pred_val.flatten() # 3. 신호 분포 시각화 fig3 = plot_signal_distribution(y_pred_val, best_threshold) if fig3: fig3.savefig(plots_dir / "signal_distribution.png", dpi=300, bbox_inches='tight') plt.close(fig3) print(" - signal_distribution.png 저장 완료") # 4. 가격 예측 시각화 fig5 = plot_price_predictions(best_model, data_dict, best_threshold, ticker_encoder) if fig5: fig5.savefig(plots_dir / "price_predictions.png", dpi=300, bbox_inches='tight') plt.close(fig5) print(" - price_predictions.png 저장 완료") # 5. 그래프 임베딩 시각화 sector_industry_df = data_dict.get('sector_industry_df') # 섹터-산업 데이터가 없으면 생성 시도 if sector_industry_df is None: try: # 티커 인코더에서 티커 목록 추출 if hasattr(ticker_encoder, 'classes_'): tickers = ticker_encoder.classes_.tolist() elif hasattr(ticker_encoder, 'mapping'): tickers = list(ticker_encoder.mapping.keys()) else: tickers = None if tickers: from ..data.hierarchical_embedding import get_industry_data sector_industry_df = get_industry_data(tickers) print(f"섹터-산업 데이터 동적 생성: {len(sector_industry_df) if sector_industry_df is not None else 0}개 종목") except Exception as e: print(f"섹터-산업 데이터 생성 실패: {e}") sector_industry_df = None if sector_industry_df is not None and len(sector_industry_df) > 0: try: # t-SNE와 PCA 시각화 생성 save_path_tsne = plots_dir / 'graph_embedding_tsne.png' save_path_pca = plots_dir / 'graph_embedding_pca.png' # 함수 호출 plot_graph_embeddings( sector_industry_df, save_path_tsne=str(save_path_tsne), save_path_pca=str(save_path_pca) ) print(" - graph_embedding_tsne.png 저장 완료") print(" - graph_embedding_pca.png 저장 완료") except Exception as e: print(f"그래프 임베딩 시각화 오류: {e}") import traceback traceback.print_exc() else: print(" - 섹터-산업 데이터가 없어 그래프 임베딩 시각화를 건너뜁니다.") except Exception as e: print(f"시각화 중 오류 발생: {e}") import traceback traceback.print_exc() print("시각화를 건너뛰고 계속 진행합니다.") # 테스트 세트 평가 test_backtest = None if all(key in data_dict for key in ['x_test', 'y_test', 'ticker_test', 'time_diffs_test']): print("\n===== 테스트 세트 성능 평가 =====") try: # 테스트 데이터 x_test = data_dict['x_test'] y_test = data_dict['y_test'] ticker_test = data_dict['ticker_test'] time_diffs_test = data_dict['time_diffs_test'] # 섹터/산업 정보 처리 sector_test = data_dict.get('sector_test', np.zeros_like(ticker_test)) industry_test = data_dict.get('industry_test', np.zeros_like(ticker_test)) # 데이터 타입 변환 ticker_test = np.asarray(ticker_test, dtype=np.int32) sector_test = np.asarray(sector_test, dtype=np.int32) industry_test = np.asarray(industry_test, dtype=np.int32) time_diffs_test = np.asarray(time_diffs_test, dtype=np.float32) # 테스트 데이터 정리 x_test_clean = clean_numeric_data(x_test, replace_nan=0.0, replace_inf=0.0, verbose=False) # 예측 수행 test_preds = best_model.predict([ tf.cast(x_test_clean, tf.float32), tf.cast(ticker_test, tf.int32), tf.cast(sector_test, tf.int32), tf.cast(industry_test, tf.int32), tf.cast(time_diffs_test, tf.float32) ], verbose=0) if isinstance(test_preds, list): y_pred_test = test_preds[0].flatten() else: y_pred_test = test_preds.flatten() # 백테스트 실행 test_backtest = backtest_by_ticker( predictions=y_pred_test, actual_returns=y_test.flatten(), ticker_ids=ticker_test.flatten(), threshold=best_threshold, commission=0.0025, risk_free_rate=risk_free_rate ) # 종목별 메트릭 계산 ticker_returns = [info['total_return'] for _, info in test_backtest['by_ticker'].items()] ticker_sharpes = [info['sharpe_ratio'] for _, info in test_backtest['by_ticker'].items()] avg_ticker_return = np.mean(ticker_returns) avg_ticker_sharpe = np.mean(ticker_sharpes) # 테스트 결과 출력 print(f"\n----- 포트폴리오 성능 -----") print(f"테스트 세트 총 수익률: {test_backtest['portfolio']['total_return']:.4f}") print(f"테스트 세트 샤프 비율: {test_backtest['portfolio']['sharpe_ratio']:.4f}") print(f"테스트 세트 최대 낙폭: {test_backtest['portfolio']['max_drawdown']:.4f}") print(f"테스트 세트 거래 수: {len(test_backtest['portfolio'].get('trades', []))}") print(f"\n----- 개별 종목 평균 성능 -----") print(f"테스트 세트 평균 종목 수익률: {avg_ticker_return:.4f}") print(f"테스트 세트 평균 종목 샤프 비율: {avg_ticker_sharpe:.4f}") except Exception as e: print(f"테스트 세트 평가 중 오류 발생: {e}") import traceback traceback.print_exc() # 결과 저장 optimization_results = { 'best_config': best_config, 'best_result': best_result, 'results': results, 'solver': 'rk4' } # 모델 및 결과 저장 if save: # 모델 저장 if best_model: # 인코더 정보 추출 encoders = None if ticker_encoder: if hasattr(ticker_encoder, 'mapping'): encoders = { 'ticker_encoder': ticker_encoder.mapping } elif hasattr(ticker_encoder, 'classes_'): encoders = { 'ticker_encoder': {i: tick for i, tick in enumerate(ticker_encoder.classes_)} } # 기본 경로 설정 models_dir = Path(get_project_root()) / "models" results_dir = models_dir / "results" results_dir.mkdir(parents=True, exist_ok=True) if model_output is None: model_output = results_dir / "best_contime_model.keras" else: model_output = Path(model_output) if not model_output.is_absolute(): model_output = results_dir / model_output.name if not str(model_output).endswith('.keras'): model_output = Path(str(model_output).replace('.h5', '') + '.keras') # 모델 저장 save_model( model=best_model, model_path=model_output, config=best_config, encoders=encoders ) # 임계값 및 성능 정보 저장 threshold_info = { 'best_threshold': best_threshold, 'config': best_config, 'avg_ticker_sharpe': best_result['metrics'].get('avg_ticker_sharpe', 0), 'portfolio_sharpe': best_result['metrics']['sharpe_ratio'], 'total_return': best_result['metrics']['total_return'], 'avg_ticker_return': float(avg_return), 'avg_ticker_win_rate': float(avg_win_rate), 'avg_ticker_mdd': float(avg_mdd), 'trade_count': best_result['metrics']['trade_count'], 'total_opportunities': best_result['total_opportunities'], 'trade_ratio': float(best_result['metrics']['trade_count'] / best_result['total_opportunities']), 'min_expected_trades': best_result['min_expected_trades'] } if test_backtest: threshold_info['test_metrics'] = { 'total_return': test_backtest['portfolio']['total_return'], 'sharpe_ratio': test_backtest['portfolio']['sharpe_ratio'], 'max_drawdown': test_backtest['portfolio']['max_drawdown'], 'trade_count': len(test_backtest['portfolio'].get('trades', [])) } # 메타데이터 저장 meta_path = models_dir / "results" / f"{model_output.stem}_meta.json" save_metadata(threshold_info, meta_path) # 결과 저장 if output_path: output_path = Path(output_path) if not output_path.is_absolute(): output_path = Path(get_project_root()) / "models" / "results" / output_path save_results(optimization_results, output_path) # 시각화 완료 메시지 if run_visualizations: print(f"\n시각화 파일들이 {plots_dir}에 저장되었습니다:") saved_plots = list(plots_dir.glob("*.png")) if saved_plots: for plot_file in saved_plots: print(f" - {plot_file.name}") else: print(" - 저장된 시각화 파일이 없습니다.") return optimization_results