171 lines
4.7 KiB
Python
171 lines
4.7 KiB
Python
"""
|
|
KAP(한국자산평가) 등급별 회사채 기준수익률 수집 모듈
|
|
|
|
KAP 웹사이트(koreaap.com)에서 공모회사채(무보증) 1년 만기 등급별
|
|
기준수익률과 국고채 수익률을 수집합니다.
|
|
|
|
데이터 출처:
|
|
- https://www.koreaap.com/kor/valuation01_01.html
|
|
- 채권금리 기준수익률 > 공모회사채 (무보증)
|
|
|
|
사용법:
|
|
ytm_data = get_ytm_data("2025-12-31")
|
|
# {'rf': 2.550, 'AAA': 2.986, 'AA+': 3.045, ...}
|
|
"""
|
|
|
|
import logging
|
|
from typing import Dict, Optional
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
# ============================================================
|
|
# 하드코딩 Fallback 데이터
|
|
# KAP 스크래핑이 불가능할 경우 사용하는 기준 데이터
|
|
# ============================================================
|
|
|
|
# 2025-12-31 기준 KAP 공모회사채(무보증) 1년 만기 기준수익률 (%)
|
|
FALLBACK_YTM_20251231: Dict[str, float] = {
|
|
"rf": 2.550, # 국고채 1Y
|
|
"AAA": 2.986,
|
|
"AA+": 3.045,
|
|
"AA": 3.094,
|
|
"AA-": 3.132,
|
|
"A+": 3.269,
|
|
"A": 3.439,
|
|
"A-": 3.704,
|
|
"BBB+": 4.749,
|
|
"BBB": 5.442,
|
|
"BBB-": 6.442,
|
|
}
|
|
|
|
|
|
def get_ytm_data(
|
|
date: str = "2025-12-31",
|
|
use_fallback: bool = True
|
|
) -> Dict[str, float]:
|
|
"""
|
|
KAP 등급별 1Y 회사채 수익률 + 국고채 수익률 반환
|
|
|
|
Parameters
|
|
----------
|
|
date : str
|
|
기준일 (YYYY-MM-DD). 기본값 2025-12-31
|
|
use_fallback : bool
|
|
True이면 하드코딩 fallback 데이터 사용 (스크래핑 없이)
|
|
|
|
Returns
|
|
-------
|
|
Dict[str, float]
|
|
{'rf': 국고채1Y, 'AAA': AAA_YTM, 'AA+': ..., ...}
|
|
단위: % (예: 2.986)
|
|
"""
|
|
if use_fallback or date == "2025-12-31":
|
|
logger.info(f"KAP YTM fallback 데이터 사용 (기준일: {date})")
|
|
return FALLBACK_YTM_20251231.copy()
|
|
|
|
# 향후 KAP 웹 스크래핑 구현 가능
|
|
# try:
|
|
# return _scrape_kap(date)
|
|
# except Exception as e:
|
|
# logger.warning(f"KAP 스크래핑 실패, fallback 사용: {e}")
|
|
# return FALLBACK_YTM_20251231.copy()
|
|
|
|
logger.info(f"KAP YTM fallback 데이터 사용 (기준일: {date})")
|
|
return FALLBACK_YTM_20251231.copy()
|
|
|
|
|
|
def compute_spreads(ytm_data: Dict[str, float]) -> Dict[str, float]:
|
|
"""
|
|
등급별 신용스프레드 계산 (= 회사채 YTM - 국고채 YTM)
|
|
|
|
Parameters
|
|
----------
|
|
ytm_data : Dict[str, float]
|
|
get_ytm_data() 반환값
|
|
|
|
Returns
|
|
-------
|
|
Dict[str, float]
|
|
등급별 스프레드 (단위: bp, 1bp = 0.01%)
|
|
예: {'AAA': 43.6, 'AA+': 49.5, ...}
|
|
"""
|
|
rf = ytm_data["rf"]
|
|
spreads = {}
|
|
for grade, ytm in ytm_data.items():
|
|
if grade == "rf":
|
|
continue
|
|
spreads[grade] = round((ytm - rf) * 100, 1) # % → bp
|
|
return spreads
|
|
|
|
|
|
# ============================================================
|
|
# 대등급 매핑
|
|
# ============================================================
|
|
|
|
# 노치 → 대등급 매핑
|
|
NOTCH_TO_BROAD = {
|
|
"AAA": "AAA",
|
|
"AA+": "AA", "AA": "AA", "AA-": "AA",
|
|
"A+": "A", "A": "A", "A-": "A",
|
|
"BBB+": "BBB", "BBB": "BBB", "BBB-": "BBB",
|
|
"BB+": "BB", "BB": "BB", "BB-": "BB",
|
|
"B+": "B", "B": "B", "B-": "B",
|
|
}
|
|
|
|
|
|
def compute_broad_grade_spreads(
|
|
spreads: Dict[str, float],
|
|
method: str = "mid"
|
|
) -> Dict[str, float]:
|
|
"""
|
|
노치등급 스프레드를 대등급(AAA, AA, A, BBB)으로 집계
|
|
|
|
Parameters
|
|
----------
|
|
spreads : Dict[str, float]
|
|
노치등급별 스프레드 (bp)
|
|
method : str
|
|
'mid': 중간 노치 값 사용 (AA+ AA AA- → AA 값)
|
|
'mean': 노치 평균
|
|
|
|
Returns
|
|
-------
|
|
Dict[str, float]
|
|
대등급별 스프레드 (bp)
|
|
"""
|
|
from collections import defaultdict
|
|
groups = defaultdict(list)
|
|
|
|
for notch, spread in spreads.items():
|
|
broad = NOTCH_TO_BROAD.get(notch, notch)
|
|
groups[broad].append(spread)
|
|
|
|
result = {}
|
|
for broad, values in groups.items():
|
|
if method == "mid":
|
|
# 중간 노치 (3개면 가운데, 1개면 그대로)
|
|
result[broad] = sorted(values)[len(values) // 2]
|
|
else:
|
|
result[broad] = sum(values) / len(values)
|
|
|
|
return result
|
|
|
|
|
|
if __name__ == "__main__":
|
|
# 테스트 실행
|
|
ytm = get_ytm_data("2025-12-31")
|
|
print("=== KAP 1Y YTM (2025-12-31) ===")
|
|
for grade, rate in ytm.items():
|
|
print(f" {grade:>5}: {rate:.3f}%")
|
|
|
|
print("\n=== 신용스프레드 (bp) ===")
|
|
spreads = compute_spreads(ytm)
|
|
for grade, sp in spreads.items():
|
|
print(f" {grade:>5}: {sp:.1f}bp")
|
|
|
|
print("\n=== 대등급 스프레드 (bp) ===")
|
|
broad = compute_broad_grade_spreads(spreads)
|
|
for grade, sp in broad.items():
|
|
print(f" {grade:>5}: {sp:.1f}bp")
|