26J29e1

"""
[2.5단계-패턴 패치] OpenCost 수집 누락 버그 방어 및 Core-Hours 기반 파레토 왜곡 보정 엔진
실행: python step2_5_fix_pareto.py
"""

import pandas as pd
import numpy as np
from pathlib import Path

# 공통 프로젝트 경로 설정 구성원 로드
from config import MERGED_DIR

def fix_opencost_time_and_recalculate_pareto():
    src_path = MERGED_DIR / "enriched_7d.parquet"
    if not src_path.exists():
        print(f"❌ 원천 enriched_7d.parquet 파일이 없습니다. 경로를 확인하세요: {src_path}")
        return

    # 1. 데이터 로드
    df = pd.read_parquet(src_path)
    print(f"📊 원천 데이터 로드 완료 (총 행수: {len(df):,}개)")

    # ── [🛡️ KeyError 방어 블록] minutes_running 컬럼 존재 여부 검증 및 강제 생성 ──
    if "minutes_running" not in df.columns:
        print("⚠️ [안내] 'minutes_running' 컬럼이 데이터프레임에 완전히 누락되었습니다. 구조 복구 시작.")
        # 만약 오픈코스트 원천 이름인 'minutes'가 살아있다면 이를 복사, 없으면 0.0으로 베이스라인 초기화
        if "minutes" in df.columns:
            df["minutes_running"] = df["minutes"].astype(float)
            print("  -> 원천 'minutes' 필드를 포착하여 'minutes_running'으로 이식했습니다.")
        else:
            df["minutes_running"] = 0.0
            print("  -> 원천 필드가 없어 'minutes_running' 컬럼을 0.0으로 신규 생성했습니다.")

    # ── [버그 패치 2] 가동 시간이 마비(전부 0 또는 NaN)되어 있다면 기술 스택별 시간 강제 동기화 ──
    if (df["minutes_running"] == 0).all() or df["minutes_running"].isna().all():
        print("🚨 [경고] 가동 시간 지표가 완전히 유실(모두 0 또는 NaN)된 상태가 확인되었습니다.")
        print("⚙️ 상주형(정적) vs 배치형(동적) 워크로드 타임라인 분리 주입 가동...")
        
        # 데이터 정합성을 위해 대문자 통일
        df['workload_type'] = df['workload_type'].str.upper()
        
        # 7일 풀 상주형 기준분 수 = 7d * 24h * 60m = 10,080분
        df["minutes_running"] = np.where(
            df['workload_type'].isin(["STARROCKS", "POSTGRESQL", "JUPYTERLAB", "AIRFLOW"]),
            10080.0, # 정적 상주형 워크로드 가동 시간 정상 복구
            np.where((df["minutes_running"] == 0) | (df["minutes_running"].isna()), 5.0, df["minutes_running"]) # Spark 등 휘발성 배치
        )
        print("✅ [시간 복구 완료] 정적 워크로드 10,080분 / 동적 워크로드 5분으로 리밸런싱 완료.")

    # ── [핵심 연산 3] 왜곡 차단을 위한 Core-Hours (코어×시간) 연산 가동 ──
    print("⚙️ 시간 가중치(Core-Hours) 계산 중...")
    
    # 타입 에러 방지를 위한 Float 강제 캐스팅
    df["cpu_request_max"] = df["cpu_request_max"].astype(float)
    df["cpu_usage_p95"] = df["cpu_usage_p95"].astype(float)
    
    # 코어-시간 연산
    df["cpu_allocated_core_hours"] = df["cpu_request_max"] * (df["minutes_running"] / 60.0)
    df["cpu_usage_core_hours"] = df["cpu_usage_p95"] * (df["minutes_running"] / 60.0)
    df["cpu_waste_core_hours"] = (df["cpu_allocated_core_hours"] - df["cpu_usage_core_hours"]).clip(lower=0)

    # 메모리 가중치 단위(GB-Hours) 보정 연산 추가 (하방 파이프라인 에러 방어)
    if "mem_request_max" in df.columns and "mem_usage_p95" in df.columns:
        df["mem_allocated_gb_hours"] = df["mem_request_max"].astype(float) * (df["minutes_running"] / 60.0)
        df["mem_usage_gb_hours"] = df["mem_usage_p95"].astype(float) * (df["minutes_running"] / 60.0)

    # 수정된 무결성 데이터 원부 세이브
    df.to_parquet(MERGED_DIR / "enriched_fixed_7d.parquet", index=False)
    print("💾 보정 완료된 데이터 원부가 enriched_fixed_7d.parquet에 저장되었습니다.")
    
    # ── [핵심 연산 4] Core-Hours 기반 파레토 재집계 및 서열화 ──
    print("📈 보정된 파레토 누적 지표 연산 중...")
    
    df_ns = df.groupby("namespace").agg(
        minutes_running_sum   = ("minutes_running", "sum"),
        container_cnt         = ("container", "count"),
        total_request_cores   = ("cpu_request_max", "sum"),
        total_waste_cores_raw = ("cpu_waste_cores", "sum"),
        total_allocated_core_hours = ("cpu_allocated_core_hours", "sum"),
        total_waste_core_hours     = ("cpu_waste_core_hours", "sum")
    ).reset_index()
    
    # 낭비가 심한 순서대로 정렬 (파레토 대전제)
    df_ns = df_ns.sort_values(by="total_waste_core_hours", ascending=False).reset_index(drop=True)
    
    global_total_waste_core_hours = df_ns["total_waste_core_hours"].sum()
    if global_total_waste_core_hours == 0:
        global_total_waste_core_hours = 0.1
        
    df_ns["waste_share_pct"] = (df_ns["total_waste_core_hours"] / global_total_waste_core_hours * 100).round(2)
    df_ns["waste_cumsum_pct"] = df_ns["waste_share_pct"].cumsum().round(2)
    df_ns["is_top_80_percent_offender"] = df_ns["waste_cumsum_pct"] <= 80.5

    # 보정본 파레토 결과 세이브
    df_ns.to_parquet(MERGED_DIR / "pareto_fixed_ns.parquet", index=False)
    
    print("\n" + "="*95)
    print("📢 [보정 완료] Core-Hours 기반 인프라 비용 누수 상위 5개 Namespaces 명세")
    print("="*95)
    print(f"{'Rank':<5}{'Namespace':<25}{'Pods':<8}{'Waste(Core-Hours)':<25}{'Share(%)':<12}{'Cumsum(%)':<12}")
    print("-"*95)
    for idx, row in df_ns.head(5).iterrows():
        print(f"{idx+1:<5}{row['namespace']:<25}{int(row['container_cnt']):<8}{row['total_waste_core_hours']:<25,.1f}{row['waste_share_pct']:<12.2f}{row['waste_cumsum_pct']:<12.2f}")
    print("="*95)

if __name__ == "__main__":
    fix_opencost_time_and_recalculate_pareto()