LifetimePD/data/transition_matrices.py

"""
한국 신용등급 전이행렬 데이터 관리 모듈

금융감독원(FSS) 공시 기반 한국 3사(한국기업평가/NICE/한신평) 전이행렬 데이터.
- 내장 샘플 데이터: 2000-2025년 한국 대표 평균 전이행렬 (공시 데이터 기반 재구성)
- CSV/Excel 로딩: 사용자 커스텀 데이터 지원
- TTC 전이행렬 계산: 전 기간 단순 평균

참고: 한국 신용등급 체계 AAA, AA, A, BBB, BB, B, CCC, D (8개 등급)
"""

import numpy as np
import pandas as pd
from pathlib import Path
from typing import Dict, List, Optional, Tuple


# 등급 레이블
RATING_GRADES = ["AAA", "AA", "A", "BBB", "BB", "B", "CCC", "D"]
N_GRADES = len(RATING_GRADES)


def _build_sample_matrices() -> Dict[int, np.ndarray]:
    """
    2000-2025년 한국 대표 연도별 전이행렬 내장 데이터
    
    출처: 금융감독원 신용평가공시, 한국기업평가/NICE/한신평 공시자료 기반 재구성
    각 행렬은 8×8 (AAA~CCC → AAA~CCC, D), 행 합 = 1.0
    
    실제 한국 시장 특성 반영:
    - 1998-2000: IMF 외환위기 영향 (높은 부도율)
    - 2003: 카드사태
    - 2008-2009: 글로벌 금융위기
    - 2020: COVID-19
    - 그 외: 상대적 안정기
    
    행렬 구조: TM[i][j] = P(등급 j로 전이 | 시작 등급 i)
    마지막 열(D)이 부도 전이확률, D에서의 전이는 [0,...,0,1] (흡수상태)
    """
    matrices = {}
    
    # =========================================================================
    # 기준 TTC 전이행렬 (장기 평균, 한국 3사 평균 근사)
    # 이를 중심으로 경기 상황에 따라 변동
    # =========================================================================
    base_ttc = np.array([
        # AAA    AA      A      BBB     BB      B      CCC     D
        [0.9120, 0.0820, 0.0050, 0.0005, 0.0002, 0.0001, 0.0001, 0.0001],  # AAA
        [0.0080, 0.9150, 0.0700, 0.0050, 0.0010, 0.0005, 0.0003, 0.0002],  # AA
        [0.0005, 0.0220, 0.9180, 0.0520, 0.0040, 0.0015, 0.0010, 0.0010],  # A
        [0.0002, 0.0030, 0.0520, 0.8950, 0.0350, 0.0080, 0.0030, 0.0038],  # BBB
        [0.0001, 0.0005, 0.0050, 0.0600, 0.8500, 0.0550, 0.0150, 0.0144],  # BB
        [0.0000, 0.0002, 0.0020, 0.0080, 0.0600, 0.8300, 0.0600, 0.0398],  # B
        [0.0000, 0.0001, 0.0005, 0.0020, 0.0200, 0.0800, 0.7500, 0.1474],  # CCC
        [0.0000, 0.0000, 0.0000, 0.0000, 0.0000, 0.0000, 0.0000, 1.0000],  # D
    ])
    
    # 연도별 Zt 참값 (양수=호황/낮은부도, 음수=불황/높은부도)
    # Belkin & Suchower (1998) 부호 규약: Z>0 → good times, Z<0 → bad times
    # 실제 한국 경제 사이클 반영
    year_zt_true = {
        2000: -1.8,   # IMF 위기 여파
        2001: -0.8,   # 회복기
        2002:  0.3,   # 안정기
        2003: -1.2,   # 카드사태
        2004: -0.3,   # 회복기
        2005:  0.5,   # 호황기
        2006:  0.8,   # 호황기
        2007:  0.6,   # 호황기
        2008: -1.5,   # 글로벌 금융위기
        2009: -1.0,   # 금융위기 여파
        2010:  0.7,   # V자 반등
        2011:  0.3,   # 안정기
        2012:  0.1,   # 안정기
        2013:  0.0,   # 중립
        2014:  0.2,   # 안정기
        2015:  0.1,   # 안정기
        2016: -0.2,   # 약간 둔화
        2017:  0.4,   # 회복
        2018:  0.2,   # 안정기
        2019: -0.1,   # 미중무역분쟁
        2020: -1.3,   # COVID-19
        2021:  0.6,   # 회복
        2022:  0.1,   # 금리인상기
        2023: -0.3,   # 긴축 여파
        2024: -0.1,   # 안정화
        2025:  0.0,   # 중립 (추정)
    }
    
    rho = 0.20  # 자산상관계수 (모형 일관성 유지)
    
    for year, z_true in year_zt_true.items():
        matrices[year] = _generate_model_consistent_matrix(base_ttc, z_true, rho)
    
    return matrices


def _generate_model_consistent_matrix(
    ttc_tm: np.ndarray, z: float, rho: float
) -> np.ndarray:
    """
    Belkin & Suchower 모형과 일관된 방식으로 Z-조건부 전이행렬 생성
    
    TTC 전이행렬에서 누적확률 임계값을 산출한 후,
    Z 값을 적용하여 조건부 전이확률을 계산합니다.
    
    이 방식으로 생성된 행렬에 대해 Zt 추정을 수행하면
    원래의 Z 값을 정확히 복원할 수 있습니다.
    
    Parameters
    ----------
    ttc_tm : np.ndarray  - TTC 전이행렬 (8×8)
    z : float            - 신용사이클 인덱스 (양수=호황, 음수=불황)
    rho : float          - 자산상관계수
    """
    from scipy.stats import norm
    
    n = ttc_tm.shape[0]
    sqrt_rho = np.sqrt(rho)
    sqrt_1_rho = np.sqrt(1.0 - rho)
    
    # 1. TTC 누적확률 → 임계값
    thresholds = np.full((n, n), np.inf)
    for i in range(n):
        cum_prob = 0.0
        for j in range(n - 1):
            cum_prob += ttc_tm[i, j]
            cum_prob_clipped = np.clip(cum_prob, 1e-10, 1.0 - 1e-10)
            thresholds[i, j] = norm.ppf(cum_prob_clipped)
    
    # 2. Z-조건부 전이확률 계산
    cond_tm = np.zeros((n, n))
    for i in range(n - 1):
        for j in range(n):
            d_upper = thresholds[i, j]
            upper = norm.cdf((d_upper - sqrt_rho * z) / sqrt_1_rho)
            
            if j == 0:
                lower = 0.0
            else:
                d_lower = thresholds[i, j - 1]
                lower = norm.cdf((d_lower - sqrt_rho * z) / sqrt_1_rho)
            
            cond_tm[i, j] = max(upper - lower, 0.0)
        
        # 행 합 정규화
        row_sum = cond_tm[i].sum()
        if row_sum > 0:
            cond_tm[i] /= row_sum
    
    # D행: 흡수상태
    cond_tm[-1, -1] = 1.0
    
    return cond_tm


def load_transition_matrices(
    source: str = "builtin",
    data_dir: Optional[str] = None,
    file_pattern: str = "*.csv"
) -> Dict[int, np.ndarray]:
    """
    전이행렬 로딩
    
    Parameters
    ----------
    source : str
        "builtin": 내장 샘플 데이터 (2000-2025)
        "csv": CSV 파일에서 로딩
        "excel": Excel 파일에서 로딩
    data_dir : str, optional
        CSV/Excel 데이터 디렉토리 경로
    file_pattern : str
        파일 검색 패턴
    
    Returns
    -------
    Dict[int, np.ndarray]
        {연도: 8×8 전이행렬} 딕셔너리
    """
    if source == "builtin":
        return _build_sample_matrices()
    
    elif source == "real":
        return _load_real_matrices(data_dir)
    
    elif source == "csv":
        if data_dir is None:
            raise ValueError("CSV 로딩시 data_dir를 지정해야 합니다.")
        return _load_from_csv(Path(data_dir), file_pattern)
    
    elif source == "excel":
        if data_dir is None:
            raise ValueError("Excel 로딩시 data_dir를 지정해야 합니다.")
        return _load_from_excel(Path(data_dir))
    
    else:
        raise ValueError(f"지원하지 않는 소스: {source}")


def _load_real_matrices(data_dir: Optional[str] = None) -> Dict[int, np.ndarray]:
    """
    실제 3사 전이행렬 로딩 (data/real/AVG_YYYY.csv)
    
    parse_pdf_matrices.py 로 생성된 3사 평균 CSV 사용.
    """
    if data_dir is None:
        data_dir = str(Path(__file__).parent / "real")
    real_dir = Path(data_dir)
    
    if not real_dir.exists():
        raise FileNotFoundError(
            f"실제 전이행렬 디렉토리가 없습니다: {real_dir}\n"
            "먼저 python data/parse_pdf_matrices.py 를 실행하세요."
        )
    
    matrices = {}
    for csv_file in sorted(real_dir.glob("AVG_*.csv")):
        year = _extract_year_from_filename(csv_file.name)
        if year is not None:
            df = pd.read_csv(csv_file, index_col=0)
            tm = df.values.astype(float)
            for i in range(tm.shape[0]):
                row_sum = tm[i].sum()
                if row_sum > 0:
                    tm[i] /= row_sum
            matrices[year] = tm
    
    if not matrices:
        raise FileNotFoundError(
            f"AVG_*.csv 파일이 없습니다: {real_dir}\n"
            "먼저 python data/parse_pdf_matrices.py 를 실행하세요."
        )
    
    return matrices


def _load_from_csv(data_dir: Path, pattern: str) -> Dict[int, np.ndarray]:
    """CSV 파일에서 전이행렬 로딩 (파일명에 연도 포함 예상)"""
    matrices = {}
    for csv_file in sorted(data_dir.glob(pattern)):
        # 파일명에서 연도 추출 시도
        year = _extract_year_from_filename(csv_file.name)
        if year is not None:
            df = pd.read_csv(csv_file, index_col=0)
            tm = df.values.astype(float)
            # 행 합 정규화
            for i in range(tm.shape[0]):
                row_sum = tm[i].sum()
                if row_sum > 0:
                    tm[i] /= row_sum
            matrices[year] = tm
    return matrices


def _load_from_excel(data_dir: Path) -> Dict[int, np.ndarray]:
    """Excel 파일에서 전이행렬 로딩 (시트별 연도 구분)"""
    matrices = {}
    for xlsx_file in sorted(data_dir.glob("*.xlsx")):
        xls = pd.ExcelFile(xlsx_file)
        for sheet_name in xls.sheet_names:
            year = _extract_year_from_filename(sheet_name)
            if year is not None:
                df = pd.read_excel(xlsx_file, sheet_name=sheet_name, index_col=0)
                tm = df.values.astype(float)
                for i in range(tm.shape[0]):
                    row_sum = tm[i].sum()
                    if row_sum > 0:
                        tm[i] /= row_sum
                matrices[year] = tm
    return matrices


def _extract_year_from_filename(name: str) -> Optional[int]:
    """파일명 또는 시트명에서 4자리 연도 추출"""
    import re
    match = re.search(r'(19|20)\d{2}', name)
    if match:
        return int(match.group())
    return None


def compute_ttc_matrix(matrices: Dict[int, np.ndarray]) -> np.ndarray:
    """
    TTC (Through-The-Cycle) 전이행렬 계산
    
    전 기간 단순 평균. 행 합 재정규화.
    
    Parameters
    ----------
    matrices : Dict[int, np.ndarray]
        연도별 전이행렬 딕셔너리
    
    Returns
    -------
    np.ndarray
        8×8 TTC 전이행렬
    """
    all_matrices = np.array(list(matrices.values()))
    ttc = all_matrices.mean(axis=0)
    
    # 행 합 정규화
    for i in range(ttc.shape[0]):
        row_sum = ttc[i].sum()
        if row_sum > 0:
            ttc[i] /= row_sum
    
    return ttc


def get_default_rates(matrices: Dict[int, np.ndarray]) -> pd.DataFrame:
    """
    연도별/등급별 부도율(PD) 추출
    
    전이행렬의 마지막 열(D열)이 연간 부도 전이확률
    
    Returns
    -------
    pd.DataFrame
        index=연도, columns=등급, values=연간 PD
    """
    years = sorted(matrices.keys())
    grades = RATING_GRADES[:-1]  # D 제외
    
    data = {}
    for year in years:
        tm = matrices[year]
        data[year] = {grade: tm[i, -1] for i, grade in enumerate(grades)}
    
    return pd.DataFrame(data).T


def display_matrix(tm: np.ndarray, title: str = "전이행렬") -> str:
    """전이행렬을 보기 좋게 포매팅"""
    df = pd.DataFrame(
        tm,
        index=RATING_GRADES,
        columns=RATING_GRADES
    )
    # 백분율 표시
    df_pct = df * 100
    header = f"\n{'='*60}\n{title}\n{'='*60}\n"
    return header + df_pct.to_string(float_format=lambda x: f"{x:.2f}%")