정상화

app.py

@@ -1,136 +1,208 @@
-# import gradio as gr
-# from gradio_leaderboard import Leaderboard, ColumnFilter, SelectColumns
-# import pandas as pd
-# from apscheduler.schedulers.background import BackgroundScheduler
-# from huggingface_hub import snapshot_download
-#
-# from src.about import (
-#     CITATION_BUTTON_LABEL,
-#     CITATION_BUTTON_TEXT,
-#     EVALUATION_QUEUE_TEXT,
-#     INTRODUCTION_TEXT,
-#     LLM_BENCHMARKS_TEXT,
-#     TITLE,
-# )
-# from src.display.css_html_js import custom_css
-# from src.display.utils import (
-#     BENCHMARK_COLS,
-#     COLS,
-#     EVAL_COLS,
-#     EVAL_TYPES,
-#     AutoEvalColumn,
-#     ModelType,
-#     fields,
-#     WeightType,
-#     Precision
-# )
-# from src.envs import API, EVAL_REQUESTS_PATH, EVAL_RESULTS_PATH, QUEUE_REPO, REPO_ID, RESULTS_REPO, TOKEN
-# from src.populate import get_evaluation_queue_df, get_leaderboard_df
-# from src.submission.submit import add_new_eval
-#
-#
-# def restart_space():
-#     API.restart_space(repo_id=REPO_ID)
-#
-# ### Space initialisation
-# try:
-#     print(EVAL_REQUESTS_PATH)
-#     snapshot_download(
-#         repo_id=QUEUE_REPO, local_dir=EVAL_REQUESTS_PATH, repo_type="dataset", tqdm_class=None, etag_timeout=30, token=TOKEN
-#     )
-# except Exception:
-#     restart_space()
-# try:
-#     print(EVAL_RESULTS_PATH)
-#     snapshot_download(
-#         repo_id=RESULTS_REPO, local_dir=EVAL_RESULTS_PATH, repo_type="dataset", tqdm_class=None, etag_timeout=30, token=TOKEN
-#     )
-# except Exception:
-#     restart_space()
-#
-#
-#
-# scheduler = BackgroundScheduler()
-# scheduler.add_job(restart_space, "interval", seconds=1800)
-# scheduler.start()
-# demo.queue(default_concurrency_limit=40).launch()
-
+import os
+import shutil
+from datetime import datetime
+from typing import List
 
 import gradio as gr
 import pandas as pd
-import os
-import shutil
-from huggingface_hub import upload_file
-from src.envs import RESULTS_REPO
-from src.grader import grade   # grader.py 안에 채점 + 리포팅 함수
+from huggingface_hub import snapshot_download, upload_file
+
+# 프로젝트 구조에 맞춰 import (src.*)
+from src.envs import RESULTS_REPO, EVAL_RESULTS_PATH
+from src.grader import grade   # grade(submission_df, team_id) -> (score_df, report_dir)
+
+# ---------------------------
+# 팀 비밀번호 매핑 (요청 반영)
+# ---------------------------
+TEAM_PWD_MAP = {
+    "02Aug#29112#h": "JN_HACK13",
+    "02Aug!86113!g": "JN_HACK14",
+    "02Aug#33114#h": "JN_HACK15",
+}
+
+# 저장 파일명 (리더보드 집계)
+HISTORY_NAME = "graded_results.csv"
+HISTORY_PATH = os.path.join(EVAL_RESULTS_PATH, HISTORY_NAME)
+
+# ---------------------------
+# 초기화: Space 재시작 대비
+# ---------------------------
+def init_cache() -> pd.DataFrame:
+    """
+    - RESULTS_REPO 스냅샷을 EVAL_RESULTS_PATH로 다운로드(동기화)
+    - graded_results.csv 로드 (없으면 빈 DF 생성)
+    """
+    os.makedirs(EVAL_RESULTS_PATH, exist_ok=True)
+    try:
+        snapshot_download(
+            repo_id=RESULTS_REPO,
+            repo_type="dataset",
+            local_dir=EVAL_RESULTS_PATH,
+            tqdm_class=None,
+            etag_timeout=30,
+            token=os.environ.get("HF_TOKEN"),
+        )
+    except Exception as e:
+        print(f"[WARN] snapshot_download failed: {e}")
+
+    if os.path.exists(HISTORY_PATH):
+        try:
+            df = pd.read_csv(HISTORY_PATH)
+        except Exception as e:
+            print(f"[WARN] failed to read {HISTORY_NAME}: {e}")
+            df = pd.DataFrame()
+    else:
+        df = pd.DataFrame()
+
+    # 최소 스키마 보정
+    for col in ["TEAM", "TIMESTAMP"]:
+        if col not in df.columns:
+            df[col] = []
+    return df
+
+
+# 전역 히스토리 상태 (앱 부팅 시 로드)
+history_state = init_cache()
+
+
+def save_and_upload_history(history_df: pd.DataFrame) -> None:
+    """
+    - 로컬 graded_results.csv 저장
+    - RESULTS_REPO에 덮어쓰기 업로드
+    """
+    history_df.to_csv(HISTORY_PATH, index=False)
+    upload_file(
+        path_or_fileobj=HISTORY_PATH,
+        path_in_repo=HISTORY_NAME,
+        repo_id=RESULTS_REPO,
+        repo_type="dataset",
+        token=os.environ.get("HF_TOKEN"),
+    )
+
 
+# ---------------------------
+# 채점 핸들러
+# ---------------------------
+def grade_csv(passwd: str, file):
+    global history_state
 
-def grade_csv(file):
+    if file is None:
+        raise gr.Error("파일 업로드가 누락되었습니다.")
+
+    team_id = TEAM_PWD_MAP.get((passwd or "").strip(), "UNKNOWN")
+    # if not team_id:
+    #     raise gr.Error("비밀번호가 올바르지 않습니다.")
+
+    # 제출 CSV 로드
     submission_df = pd.read_csv(file.name)
 
     # 채점 및 리포트 생성
-    score_df, report_dir = grade(submission_df, team_id="team")
+    score_df, report_dir = grade(submission_df, team_id=team_id)
 
-    # 점수 CSV 저장
-    output_file = "graded_results.csv"
-    score_df.to_csv(output_file, index=False)
+    # 메타 속성 부여
+    ts = datetime.utcnow().strftime("%Y-%m-%d %H:%M:%SZ")
+    score_df.insert(0, "TEAM", team_id)
+    score_df.insert(1, "TIMESTAMP", ts)
 
-    # Hugging Face Hub 업로드
-    upload_file(
-        path_or_fileobj=output_file,
-        path_in_repo=output_file,
-        repo_id=RESULTS_REPO,
-        repo_type="dataset",
-        token=os.environ.get("HF_TOKEN"),
-    )
+    # 기존 히스토리 로드(없으면 빈 DF)
+    if os.path.exists(HISTORY_PATH):
+        try:
+            saved_df = pd.read_csv(HISTORY_PATH)
+        except Exception:
+            saved_df = pd.DataFrame()
+    else:
+        saved_df = pd.DataFrame()
+
+    # 최소 스키마 통일
+    for col in set(["TEAM", "TIMESTAMP"]) - set(saved_df.columns):
+        saved_df[col] = []
+
+    # 히스토리 append
+    merged_score = pd.concat([saved_df, score_df], ignore_index=True)
+    history_state = merged_score.copy()  # 전역 상태 갱신
+
+    # 로컬 저장 + Hub 업로드
+    save_and_upload_history(history_state)
 
     # 리포트 ZIP 생성
-    report_zip = "report.zip"
-    shutil.make_archive("report", 'zip', report_dir)
+    report_zip = f"report_{team_id}.zip"
+    shutil.make_archive(f"report_{team_id}", "zip", report_dir)
 
-    # 그래프 이미지 파일 목록 (report_dir 안에 *.png)
-    image_files = [os.path.join(report_dir, f) for f in os.listdir(report_dir) if f.endswith(".png")]
+    # 갤러리 이미지 파일 목록
+    image_files: List[str] = [
+        os.path.join(report_dir, f)
+        for f in os.listdir(report_dir)
+        if f.lower().endswith(".png")
+    ]
 
-    return score_df, report_zip, image_files
+    # UI 반환: 점수표, ZIP, 갤러리, 리더보드 갱신본
+    return score_df, report_zip, image_files, history_state
 
 
+def refresh_leaderboard():
+    """사용자 요청 시 Hub 재동기화 후 리더보드 갱신."""
+    global history_state
+    history_state = init_cache()
+    return history_state
+
+
+# ---------------------------
+# UI
+# ---------------------------
 
 with gr.Blocks() as demo:
-    gr.Markdown("## Hackathon CSV 채점기 + 리포트 생성기")
+    gr.Markdown("## Hackathon CSV 채점기 + 리포트 생성기 (재시작 안전화)")
 
     with gr.Tabs():
         with gr.Tab("평가 요청"):
             with gr.Row():
                 with gr.Column():
-                    with gr.Row():
-                        csv_input = gr.File(label="CSV 업로드", file_types=[".csv"])
-                    with gr.Row():
-                        password = gr.Textbox(label="Jupyter Notebook 비밀번호")
-                    with gr.Row():
-                        submit_button = gr.Button("평가 요청")
-                    with gr.Row():
-                        df_output = gr.Dataframe(label="평가 지표 결과")
+                    csv_input = gr.File(label="CSV 업로드", file_types=[".csv"])
+                    password = gr.Textbox(label="팀 비밀번호", type="password", placeholder="팀별로 공지된 비밀번호")
+                    submit_button = gr.Button("평가 요청", variant="primary")
+                    df_output = gr.Dataframe(label="평가 지표 결과")
 
                 with gr.Column():
-                    report_output = gr.File(label="리포트 ZIP 다운로드")
+                    report_output = gr.File(label="리포트 ZIP 다운로드", height="100px")
                     image_gallery = gr.Gallery(
                         label="Plant별 비교 그래프",
                         show_label=True,
-                        height="auto"  # 높이는 자동
+                        height="auto"
                     )
 
             # 업로드 → 채점 실행
             submit_button.click(
                 fn=grade_csv,
-                inputs=csv_input,
-                outputs=[df_output, report_output, image_gallery]
+                inputs=[password, csv_input],
+                outputs=[df_output, report_output, image_gallery],  # 마지막은 리더보드 탭의 테이블에 연결
+                queue=True
             )
 
         with gr.Tab("리더보드"):
             with gr.Row():
-                gr.Textbox("ASDF")
+                leaderboard_table = gr.Dataframe(
+                    value=history_state,
+                    label="graded_results.csv (집계)",
+                    interactive=False
+                )
+            with gr.Row():
+                refresh_btn = gr.Button("리더보드 새로고침 (Hub 동기화)")
+
+            # 새로고침: Hub → 로컬 재동기화 후 테이블 갱신
+            refresh_btn.click(
+                fn=refresh_leaderboard,
+                inputs=None,
+                outputs=leaderboard_table
+            )
+
+    # submit_button.click의 마지막 output을 리더보드 테이블에 실제 연결
+    submit_button.click(
+        fn=lambda: history_state,
+        inputs=None,
+        outputs=leaderboard_table
+    )
 
 if __name__ == "__main__":
-    # demo.launch()
-    demo.launch(debug=True, show_error=True)
+    demo.launch()
+# demo.launch(debug=True, show_error=True, enable_monitoring=True)
 

graded_results.csv

@@ -1,2 +1,3 @@
-TEAM_ID,RMSE_AC,RMSE_AC_SCALED,NMAE_RANGE,NMAE_MEAN
-13,493.28674800238497,0.35831099772441877,0.0,0.0
+TEAM,RMSE_AC,RMSE_AC_SCALED,NMAE_RANGE,NMAE_MEAN,TOTAL,TEAM_ID
+UNKNOWN,0.0,0.0,0.0,0.0,"{'rmse': 100.0, 'nmae': np.float64(100.0), 'final': np.float64(100.0)}",
+,493.28674800238497,0.3583109977244187,0.0,0.0,,13.0

report.zip

@@ -1,3 +1,3 @@
 version https://git-lfs.github.com/spec/v1
-oid sha256:108cf615dc3b4efc523a07f7568036674027ba1d31170af8c6455612e2e71d35
-size 175825
+oid sha256:cce5e403b7769f8611cc325afee0f004822ef3489e8a578c9ce7f6649bc07c1d
+size 89994

report_UNKNOWN.zip

@@ -0,0 +1,3 @@
+version https://git-lfs.github.com/spec/v1
+oid sha256:204fe86b110076e34696b62ee7cbaecf4a712703bb4c0108eb28fcd87bdcb759
+size 223248

src/grader.py

@@ -1,89 +1,225 @@
 # grader.py
 import os
+import math
 import pandas as pd
 import matplotlib.pyplot as plt
+from typing import Tuple
 from sklearn.preprocessing import MinMaxScaler
 from sklearn.metrics import mean_absolute_error, root_mean_squared_error
 
-ANSWER_PATH = "answer.csv"  # Space에 같이 넣은 정답 파일
+from src.utils import evaluate_model  # TOTAL 스코어 계산 함수
 
+ANSWER_PATH = "answer.csv"  # Space에 같이 넣은 정답 파일
 
-def grade(submission_df: pd.DataFrame, team_id: str = "submission"):
-    # 정답 불러오기
+def _safe_minmax(series: pd.Series) -> pd.Series:
+    """모든 값이 같거나 전부 NaN인 경우에도 0으로 안전 스케일링."""
+    s = series.astype(float)
+    if s.notna().sum() == 0:
+        return pd.Series([float("nan")] * len(s), index=s.index)
+    val_min = s.min()
+    val_max = s.max()
+    if pd.isna(val_min) or pd.isna(val_max) or val_min == val_max:
+        # range가 0이면 전부 0.0으로(차이 정보 없음)
+        return pd.Series([0.0 if not pd.isna(v) else float("nan") for v in s], index=s.index)
+    return (s - val_min) / (val_max - val_min)
+
+def _safe_rmse(y_true: pd.Series, y_pred: pd.Series) -> float:
+    """NaN/무한대 방어 RMSE."""
+    df = pd.concat([y_true, y_pred], axis=1).dropna()
+    if df.shape[0] == 0:
+        return float("nan")
+    a = df.iloc[:, 0].astype(float)
+    b = df.iloc[:, 1].astype(float)
+    try:
+        return root_mean_squared_error(a, b)
+    except Exception:
+        return float("nan")
+
+def _safe_nmae(y_true: pd.Series, y_pred: pd.Series, mode: str = "range") -> float:
+    """
+    mode='range' -> MAE / (max(y_true) - min(y_true))
+    mode='mean'  -> MAE / mean(y_true)
+    분모가 0/NaN이면 NaN 반환.
+    """
+    df = pd.concat([y_true, y_pred], axis=1).dropna()
+    if df.shape[0] == 0:
+        return float("nan")
+    a = df.iloc[:, 0].astype(float)
+    b = df.iloc[:, 1].astype(float)
+    try:
+        mae = mean_absolute_error(a, b)
+        if mode == "range":
+            denom = a.max() - a.min()
+        else:
+            denom = a.mean()
+        if denom is None or pd.isna(denom) or denom == 0:
+            return float("nan")
+        return mae / denom
+    except Exception:
+        return float("nan")
+
+def _plot_series(idx, y1, y2, title, ylabel, out_path):
+    plt.figure(figsize=(10, 5))
+    plt.plot(idx, y1, label="Submission")
+    plt.plot(idx, y2, label="Answer")
+    plt.xlabel("Index")
+    plt.ylabel(ylabel)
+    plt.title(title)
+    plt.legend()
+    plt.tight_layout()
+    plt.savefig(out_path)
+    plt.close()
+
+def grade(submission_df: pd.DataFrame, team_id: str = "submission") -> Tuple[pd.DataFrame, str]:
+    """
+    입력: 사용자가 업로드한 CSV DataFrame
+    출력: (score_df, report_dir)
+    - score_df: RMSE/NMAE/TOTAL 지표 1-row
+    - report_dir: 그래프 PNG들이 저장된 폴더 경로
+    """
+    # --------------------------
+    # 1) 정답/제출 정규화 & 머지
+    # --------------------------
     answer = pd.read_csv(ANSWER_PATH)
-    answer = answer[['DATE_TIME', 'PLANT_ID', 'SOURCE_KEY', 'DC_POWER','AC_POWER','DAILY_YIELD']]
+    answer = answer[['DATE_TIME', 'PLANT_ID', 'SOURCE_KEY', 'DC_POWER', 'AC_POWER', 'DAILY_YIELD']]
     answer = answer.rename(columns={
         'SOURCE_KEY': 'INVERTER_ID',
-        'DC_POWER': 'ANS_DC_POWER',
-        'AC_POWER': 'ANS_AC_POWER',
+        'DC_POWER'  : 'ANS_DC_POWER',
+        'AC_POWER'  : 'ANS_AC_POWER',
         'DAILY_YIELD': 'ANS_DAILY_YIELD'
     })
-    if 'SOURCE_KEY' in submission_df.columns:
-        submission_df = submission_df.rename(columns={"SOURCE_KEY": "INVERTER_ID"})
 
-    # merge
+    # 제출 컬럼 보정
+    sub = submission_df.copy()
+    if 'SOURCE_KEY' in sub.columns and 'INVERTER_ID' not in sub.columns:
+        sub = sub.rename(columns={"SOURCE_KEY": "INVERTER_ID"})
+
+    # 타입/정렬 보정
+    for c in ['PLANT_ID', 'INVERTER_ID']:
+        if c in sub.columns:
+            sub[c] = sub[c].astype(str)
+    for c in ['PLANT_ID', 'INVERTER_ID']:
+        if c in answer.columns:
+            answer[c] = answer[c].astype(str)
+
+    # 날짜 파싱 (불가 시 원문 유지)
+    for df_ in (answer, sub):
+        if 'DATE_TIME' in df_.columns:
+            try:
+                df_['DATE_TIME'] = pd.to_datetime(df_['DATE_TIME'])
+            except Exception:
+                pass
+
     merged_df = pd.merge(
-        answer, submission_df,
+        answer, sub,
         on=['DATE_TIME', 'PLANT_ID', 'INVERTER_ID'],
-        how='left'
-    ).sort_values(by=['DATE_TIME', 'PLANT_ID', 'INVERTER_ID'])
-
-
-    # scaler
-    scaler = MinMaxScaler()
-    merged_df['AC_POWER_SCALED'] = scaler.fit_transform(merged_df[['AC_POWER']])
-    merged_df['ANS_AC_POWER_SCALED'] = scaler.fit_transform(merged_df[['ANS_AC_POWER']])
-
-    # 지표 계산
-    rmse_ac = root_mean_squared_error(merged_df['AC_POWER'], merged_df['ANS_AC_POWER'])
-    rmse_ac_scaled = root_mean_squared_error(merged_df['AC_POWER_SCALED'], merged_df['ANS_AC_POWER_SCALED'])
+        how='left',
+        suffixes=('', '_SUB')
+    ).sort_values(by=['DATE_TIME', 'PLANT_ID', 'INVERTER_ID']).reset_index(drop=True)
+
+    # --------------------------
+    # 2) 스케일링 & 지표 계산
+    # --------------------------
+    # 원본 값
+    y_true_ac = merged_df.get('ANS_AC_POWER')
+    y_pred_ac = merged_df.get('AC_POWER')
+
+    # 스케일드
+    merged_df['AC_POWER_SCALED'] = _safe_minmax(merged_df.get('AC_POWER'))
+    merged_df['ANS_AC_POWER_SCALED'] = _safe_minmax(merged_df.get('ANS_AC_POWER'))
+
+    rmse_ac = _safe_rmse(y_pred_ac, y_true_ac)
+    rmse_ac_scaled = _safe_rmse(merged_df['AC_POWER_SCALED'], merged_df['ANS_AC_POWER_SCALED'])
+
+    # DAILY_YIELD
+    nmae_range, nmae_mean = float("nan"), float("nan")
+    if 'DAILY_YIELD' in merged_df.columns and 'ANS_DAILY_YIELD' in merged_df.columns:
+        nmae_range = _safe_nmae(merged_df['ANS_DAILY_YIELD'], merged_df['DAILY_YIELD'], mode="range")
+        nmae_mean  = _safe_nmae(merged_df['ANS_DAILY_YIELD'], merged_df['DAILY_YIELD'], mode="mean")
+
+    # TOTAL 점수 (evaluate_model의 기대 입력에 맞춤)
+    rmse_for_total = rmse_ac if not (pd.isna(rmse_ac) or math.isinf(rmse_ac)) else None
+    nmae_for_total = nmae_range if not (pd.isna(nmae_range) or math.isinf(nmae_range)) else None
+    try:
+        total = evaluate_model(rmse_for_total, nmae_for_total)
+    except Exception:
+        total = float("nan")
 
     metrics = {
+        # app.py에서 team_id/timestamp를 앞단에 삽입하므로, grader는 지표만 책임지게 구성.
         "RMSE_AC": rmse_ac,
         "RMSE_AC_SCALED": rmse_ac_scaled,
-        "NMAE_RANGE": 0.0,
-        "NMAE_MEAN": 0.0,
+        "NMAE_RANGE": nmae_range,
+        "NMAE_MEAN": nmae_mean,
+        "TOTAL": total,
     }
-
-    if 'DAILY_YIELD' in merged_df.columns:
-        mae_daily = mean_absolute_error(merged_df['ANS_DAILY_YIELD'], merged_df['DAILY_YIELD'])
-        daily_range = merged_df['ANS_DAILY_YIELD'].max() - merged_df['ANS_DAILY_YIELD'].min()
-        daily_mean = merged_df['ANS_DAILY_YIELD'].mean()
-
-        metrics["NMAE_RANGE"] = mae_daily / daily_range if daily_range != 0 else None
-        metrics["NMAE_MEAN"] = mae_daily / daily_mean if daily_mean != 0 else None
-
-    # 리포트용 결과 DF
     score_df = pd.DataFrame([metrics])
 
-    # =========================
-    # 리포팅: Plant별 그래프 저장
-    # =========================
+    # --------------------------
+    # 3) 리포팅 (그래프 PNG 저장)
+    # --------------------------
     output_dir = f"output/{team_id}"
     os.makedirs(output_dir, exist_ok=True)
 
-    for plant_id in merged_df['PLANT_ID'].unique():
-        plt.figure(figsize=(10, 5))
-        plant_data = merged_df[merged_df['PLANT_ID'] == plant_id]
-
-        plt.plot(
-            list(range(len(plant_data['DATE_TIME']))),
-            plant_data['AC_POWER'],
-            label="Submission AC_POWER"
+    # 인덱스: 동일한 길이의 정수 인덱스로 시각화(축 겹침 최소화)
+    merged_df = merged_df.reset_index(drop=True)
+    idx = list(range(len(merged_df)))
+
+    # (A) AC_POWER 원본 비교
+    try:
+        _plot_series(
+            idx,
+            merged_df['AC_POWER'],
+            merged_df['ANS_AC_POWER'],
+            title="AC_POWER Comparison (Raw)",
+            ylabel="AC Power",
+            out_path=f"{output_dir}/ac_power_raw.png",
         )
-        plt.plot(
-            list(range(len(plant_data['DATE_TIME']))),
-            plant_data['ANS_AC_POWER'],
-            label="Answer AC_POWER"
+    except Exception:
+        pass
+
+    # (B) AC_POWER 스케일드 비교
+    try:
+        _plot_series(
+            idx,
+            merged_df['AC_POWER_SCALED'],
+            merged_df['ANS_AC_POWER_SCALED'],
+            title="AC_POWER Comparison (Scaled 0-1)",
+            ylabel="Scaled AC Power",
+            out_path=f"{output_dir}/ac_power_scaled.png",
         )
-
-        plt.xlabel('Index')
-        plt.ylabel('AC Power')
-        plt.title(f"Plant {plant_id} - AC_POWER Comparison")
-        plt.legend()
-        plt.tight_layout()
-        plt.savefig(f"{output_dir}/ac_power_{plant_id}.png")
-        plt.close()
+    except Exception:
+        pass
+
+    # (C) Plant 단위 원본 비교 (원 요청 유지)
+    try:
+        for plant_id in merged_df['PLANT_ID'].dropna().unique():
+            plant_data = merged_df[merged_df['PLANT_ID'] == plant_id].reset_index(drop=True)
+            pidx = list(range(len(plant_data)))
+            _plot_series(
+                pidx,
+                plant_data['AC_POWER'],
+                plant_data['ANS_AC_POWER'],
+                title=f"Plant {plant_id} - AC_POWER Comparison",
+                ylabel="AC Power",
+                out_path=f"{output_dir}/ac_power_{plant_id}.png",
+            )
+    except Exception:
+        pass
+
+    # (D) DAILY_YIELD 비교(존재 시)
+    if 'DAILY_YIELD' in merged_df.columns and 'ANS_DAILY_YIELD' in merged_df.columns:
+        try:
+            _plot_series(
+                idx,
+                merged_df['DAILY_YIELD'],
+                merged_df['ANS_DAILY_YIELD'],
+                title="DAILY_YIELD Comparison",
+                ylabel="Daily Yield",
+                out_path=f"{output_dir}/daily_yield.png",
+            )
+        except Exception:
+            pass
 
     return score_df, output_dir
 

src/utils.py

@@ -0,0 +1,44 @@
+def evaluate_model(rmse, nmae=None, rmse_max=500, nmae_max=1.0):
+    """
+    모델 성능 점수화 함수
+    - RMSE와 NMAE는 작을수록 좋음
+    - RMSE는 [0, rmse_max], NMAE는 [0, nmae_max] 구간으로 정규화
+    - NMAE가 없는 경우 0점 처리
+    - 최종 점수는 RMSE와 NMAE 점수의 단순 평균
+
+    Args:
+      rmse : float
+          모델의 RMSE 값
+      nmae : float or None
+          모델의 NMAE 값 (없으면 None)
+      rmse_max : float
+          RMSE 최대 기준값 (default=500)
+      nmae_max : float
+          NMAE 최대 기준값 (default=1.0)
+
+    Return:
+      dict : {
+          "rmse_score": float,
+          "nmae_score": float,
+          "final_score": float
+      }
+
+    """
+    # RMSE 점수 (0~100)
+    rmse_score = max(0, (1 - rmse / rmse_max) * 100)
+
+    # NMAE 점수 (0~100)
+    if nmae is not None:
+        nmae_score = max(0, (1 - nmae / nmae_max) * 100)
+    else:
+        nmae_score = 0
+
+    # 최종 점수 = 단순 평균
+    final_score = (rmse_score + nmae_score) / 2
+
+    return final_score
+    # return {
+    #     "rmse": round(rmse_score, 2),
+    #     "nmae": round(nmae_score, 2),
+    #     "final": round(final_score, 2)
+    # }