scripts/hpo_result_analyzer_universal.py

# 필요한 라이브러리 임포트 및 자동 설치
import os
import subprocess
import sys
import logging
import pandas as pd
import argparse

def install_package(package_name):
    """지정된 패키지가 없으면 pip를 사용하여 설치합니다."""
    try:
        __import__(package_name)
    except ImportError:
        logging.info(f"{package_name} 라이브러리를 찾을 수 없어 설치합니다...")
        try:
            subprocess.check_call([sys.executable, "-m", "pip", "install", package_name, "--quiet"])
            logging.info(f"{package_name} 라이브러리 설치 성공.")
        except subprocess.CalledProcessError as e:
            logging.error(f"오류: {package_name} 라이브러리 설치 실패.")
            sys.exit(f"{package_name} 설치 실패로 스크립트를 종료합니다.")

# --- 패키지 설치 ---
required_packages = ["optuna", "pandas", "plotly"]
for pkg in required_packages:
    install_package(pkg)

import optuna

# --- 1. 경로 및 로깅 설정 ---
PROJECT_BASE_DIR = "/home/work/baro/sillok25060103"
LOG_DIR = os.path.join(PROJECT_BASE_DIR, "logs")
os.makedirs(LOG_DIR, exist_ok=True)
LOG_FILE_PATH = os.path.join(LOG_DIR, "hpo_universal_analysis.log")

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s", 
    handlers=[
        logging.FileHandler(LOG_FILE_PATH, mode='w', encoding='utf-8'), 
        logging.StreamHandler(sys.stdout) 
    ]
)

def analyze_hpo_results(db_name, study_name, output_dir, file_prefix):
    """
    지정된 Optuna DB와 Study를 분석하고 시각화 결과를 저장합니다.
    """
    logging.info("="*60)
    logging.info(f"'{study_name}' 연구 분석 시작")
    logging.info(f"DB 파일: {db_name}")
    logging.info("="*60)

    db_path = os.path.join(PROJECT_BASE_DIR, "optuna_db", db_name)
    storage_name = f"sqlite:///{db_path}"
    
    os.makedirs(output_dir, exist_ok=True)
    logging.info(f"분석 결과물(그래프) 저장 경로: {output_dir}")

    try:
        study = optuna.load_study(study_name=study_name, storage=storage_name)
        logging.info(f"연구를 성공적으로 불러왔습니다. (총 Trial: {len(study.trials)})")
    except Exception as e:
        logging.error(f"연구를 불러오는 중 오류 발생: {e}")
        return

    # --- 결과 요약 테이블 출력 ---
    logging.info("\n--- 상위 5개 Trial 요약 (eval_loss 기준) ---")
    
    # *** 수정된 부분: DataFrame을 수동으로 생성하여 안정성 확보 ***
    completed_trials = [t for t in study.trials if t.state == optuna.trial.TrialState.COMPLETE]
    if not completed_trials:
        logging.warning("완료된 Trial이 없습니다. 분석을 종료합니다.")
        return

    records = []
    for t in completed_trials:
        # 3단계 HPO처럼 user_attrs에 파라미터를 저장했을 경우와 일반적인 경우 모두 처리
        params = t.user_attrs.get('predefined_params', t.params)
        record = {
            'number': t.number,
            'eval_loss': t.value,
            'duration': t.duration,
            'params': params
        }
        records.append(record)

    df_results = pd.DataFrame(records).sort_values(by='eval_loss', ascending=True)

    with pd.option_context('display.max_rows', 5, 'display.max_columns', None, 'display.width', 120, 'display.max_colwidth', None):
        print(df_results.head())

    # --- 시각화 그래프 생성 및 저장 ---
    logging.info("\n--- 결과 시각화 그래프 생성 중 ---")
    try:
        fig_importance = optuna.visualization.plot_param_importances(study)
        importance_path = os.path.join(output_dir, f"{file_prefix}param_importances.html")
        fig_importance.write_html(importance_path)
        logging.info(f"1. 파라미터 중요도 그래프 저장 완료: {importance_path}")

        fig_history = optuna.visualization.plot_optimization_history(study)
        history_path = os.path.join(output_dir, f"{file_prefix}optimization_history.html")
        fig_history.write_html(history_path)
        logging.info(f"2. 최적화 과정 그래프 저장 완료: {history_path}")

        fig_slice = optuna.visualization.plot_slice(study)
        slice_path = os.path.join(output_dir, f"{file_prefix}slice_plot.html")
        fig_slice.write_html(slice_path)
        logging.info(f"3. 개별 파라미터 관계도 그래프 저장 완료: {slice_path}")
    except Exception as e:
        logging.error(f"시각화 그래프 생성 중 오류 발생: {e}")
    
    logging.info(f"\n'{study_name}' 연구 분석 완료.\n")


if __name__ == '__main__':
    # 터미널에서 실행할 때 인자(argument)를 받아 처리합니다.
    parser = argparse.ArgumentParser(description="Optuna HPO 결과를 분석하고 시각화합니다.")
    parser.add_argument("--db_name", type=str, required=True, help="분석할 Optuna DB 파일명 (optuna_db 폴더 내 위치)")
    parser.add_argument("--study_name", type=str, required=True, help="분석할 Study의 이름")
    parser.add_argument("--output_dir", type=str, default=os.path.join(PROJECT_BASE_DIR, "hpo_analysis_results"), help="시각화 결과물을 저장할 디렉토리")
    parser.add_argument("--file_prefix", type=str, default="analysis_", help="생성될 HTML 파일명의 접두사 (예: 'stage2_')")
    
    args = parser.parse_args()

    analyze_hpo_results(
        db_name=args.db_name,
        study_name=args.study_name,
        output_dir=args.output_dir,
        file_prefix=args.file_prefix
    )