Variet/edf
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: initial project setup - Merton-KMV model, data pipeline, .agents workflows

Browse Source
This commit is contained in:
EDF Agent
2026-03-11 19:59:38 +09:00
commit a20a7207c4
28 changed files with 3212 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

54
.agents/AGENT.md Normal file
View File

@@ -0,0 +1,54 @@
---
description: 모든 작업에 자동 적용되는 에이전트 행동 규칙. 새 대화 시작 시 반드시 이 파일을 먼저 읽습니다.
---
# Agent Rules
## Identity
당신은 이 프로젝트의 시니어 개발자입니다. 지시를 정확히 따르고, 추측보다 근거를 우선합니다.
## NEVER (절대 금지)
1. NEVER start coding without reading relevant reference documents in `.agents/references/`
2. NEVER guess when documentation exists — always check `.agents/references/` first
3. NEVER repeat a failed approach — check `.agents/references/known-issues.md` first
4. NEVER call APIs directly when helper scripts exist in `.agents/workflows/helpers/`
5. NEVER skip the pre-task checklist defined in `.agents/workflows/pre-task.md`
6. NEVER attempt the same failed approach more than 2 times
7. NEVER truncate error messages — always show the full error output
## ALWAYS (필수)
1. ALWAYS run `.agents/workflows/pre-task.md` before any implementation task
2. ALWAYS check `.agents/references/known-issues.md` before debugging
3. ALWAYS cite which reference document you consulted and what you learned
4. ALWAYS stop and ask the user if 2 consecutive attempts on the same approach fail
5. ALWAYS use existing helper scripts instead of raw API calls
6. ALWAYS read related existing code (minimum 3 files) before writing new code
## Failure Protocol
```
1st failure → Re-read reference docs → Try DIFFERENT approach
2nd failure (same issue) → STOP → Report diagnosis to user with:
- What was tried
- What failed
- Root cause hypothesis
- Suggested next steps
3rd attempt on same approach → FORBIDDEN
```
## Reference Loading Order
1. `.agents/AGENT.md` (this file — behavior rules)
2. `.agents/references/known-issues.md` (past failure patterns)
3. `.agents/references/` (project-specific knowledge)
4. `.agents/workflows/services.md` (service credentials & protocols)
5. `.agents/workflows/` (action procedures)
## PowerShell Notes
- `curl` → PowerShell에서 `Invoke-WebRequest` 별칭. **반드시 `curl.exe`** 사용
- `npm` → 실행 정책 문제 시 `cmd /c npm` 사용
- JSON 처리 시 `.py` 스크립트 권장 (PowerShell 이스케이핑 이슈 방지)

47
.agents/GUIDE.md Normal file
View File

@@ -0,0 +1,47 @@
# AI 에이전트 워크플로우 시스템 가이드
> 이 가이드는 AI 코딩 에이전트가 더 똑똑하게 동작하도록 설계된 범용 워크플로우 시스템의 사용법을 설명합니다.
---
## 파일 구조 개요
```
.agents/
├── AGENT.md ← 🧠 에이전트 헌법 (NEVER/ALWAYS 규칙)
├── GUIDE.md ← 📖 이 가이드
├── references/ ← 📚 프로젝트 지식 베이스
│ ├── architecture.md ← 아키텍처 설명
│ ├── tech-stack.md ← 기술 스택 & 버전
│ ├── conventions.md ← 코딩 컨벤션
│ └── known-issues.md ← 🔴 과거 실패 기록 (핵심!)
└── workflows/ ← ⚙️ 행동 절차
├── start.md ← 세션 시작 (룰 로딩 + devlog 복구)
├── end.md ← 세션 종료 (devlog + known-issues + Vikunja + Git)
├── pre-task.md ← 작업 전 필수 체크리스트
├── debug.md ← 디버깅 전용 절차
├── services.md ← 서비스 연동 정보 (API 토큰 포함)
├── check-gitea.md ← Gitea 현황 조회
├── check-vikunja.md ← Vikunja 태스크 조회
└── helpers/
├── vikunja_helper.py ← Vikunja API 안전 래퍼
└── wiki_helper.py ← Gitea Wiki 래퍼
```
## 각 파일의 역할
### 🧠 `AGENT.md` — 에이전트 헌법
에이전트가 **모든 대화에서 따라야 하는 글로벌 규칙**입니다.
### 📋 `pre-task.md` — 사전 점검 체크리스트
모든 구현 작업 전에 실행하는 **4단계 체크리스트**.
### 🔴 `known-issues.md` — 과거 실패 기록
**가장 중요한 파일.** 에이전트가 같은 실수를 반복하지 않도록 실패를 기록합니다.
### 🔧 `debug.md` — 디버깅 전용 워크플로우
추측 기반 디버깅을 금지하는 5단계 절차.
### 📓 Devlog — 세션별 작업 기록 (start.md / end.md에서 관리)
- **Index** (`docs/devlog/YYYY-MM-DD.md`): 매 작업마다 1줄 (필수)
- **Entry** (`docs/devlog/entries/YYYYMMDD-NNN.md`): 설계 결정/미완료/삽질 시만 (선택)

34
.agents/references/architecture.md Normal file
View File

@@ -0,0 +1,34 @@
# Architecture — EDF 프로젝트 구조
## 개요
주식 변동성 기반 한국 등급별 부도율 산출 (Merton-KMV 모형)
## 디렉토리 구조
```
EDF/
├── .agents/ # 에이전트 워크플로우 시스템
├── config/
│ └── settings.yaml # API 키, 모델 파라미터
├── src/
│ ├── data/
│ │ ├── krx_fetcher.py # KRX 주가/시총/변동성 수집
│ │ └── dart_fetcher.py # DART 재무제표 수집
│ ├── models/
│ │ └── merton.py # Merton DD/EDF 산출
│ ├── calibration/ # 보정, 블렌딩 (미구현)
│ └── validation/ # 백테스팅, 검증 (미구현)
├── data/ # 수집된 데이터
├── outputs/ # 결과물
├── docs/
│ ├── technical_methodology.md # 기술 문서
│ └── devlog/ # 세션별 작업 기록
└── requirements.txt
```
## 핵심 파이프라인
```
KRX 주가 → 변동성 계산 → Merton 풀이 → DD/EDF → Shadow Rating → 등급별 부도율
```

26
.agents/references/conventions.md Normal file
View File

@@ -0,0 +1,26 @@
# Conventions — EDF 프로젝트 코딩 컨벤션
## Python 스타일
- PEP 8 준수
- Type hints 사용 권장
- Docstring: Google style
## 커밋 메시지
```
<type>(<scope>): <description>
type: feat|fix|refactor|test|docs|chore|ci|infra
scope: data|model|calibration|validation (선택)
```
## 파일 네이밍
- 모듈: `snake_case.py`
- 클래스: `PascalCase`
- 함수: `snake_case`
- 상수: `SCREAMING_SNAKE_CASE`
## 데이터 처리
- 금액 단위: 원화 (원 단위 그대로, 변환하지 않음)
- 날짜 형식: `YYYYMMDD` (KRX/DART 호환)
- 변동성: 연환산 (annualized)
- NaN 처리: `np.nan` 사용, 0 또는 빈 문자열로 대체하지 않음

8
.agents/references/known-issues.md Normal file
View File

@@ -0,0 +1,8 @@
# Known Issues — 과거 실패 기록
> 이 파일은 에이전트가 같은 실수를 반복하지 않도록 실패를 기록합니다.
> 세션 종료 시 자동으로 새 이슈를 추가합니다.
---
(아직 기록된 이슈가 없습니다.)

31
.agents/references/tech-stack.md Normal file
View File

@@ -0,0 +1,31 @@
# Tech Stack — EDF 프로젝트 기술 스택
## 언어
- **Python 3.10+** (miniforge3/envs/edf)
## 핵심 패키지
| 패키지 | 용도 |
|--------|------|
| `numpy`, `pandas` | 데이터 처리 |
| `scipy` | Merton 비선형 방정식 풀이 (fsolve) |
| `statsmodels` | Ordered Probit (Shadow Rating) |
| `pykrx` | KRX 주가/시총 수집 |
| `opendart-reader` | DART 재무제표 API |
| `arch` | GARCH 변동성 모형 |
| `scikit-learn` | ML 보조 모형 |
| `matplotlib`, `plotly`, `seaborn` | 시각화 |
| `pyyaml` | 설정 파일 |
| `tqdm` | 프로그레스 바 |
## 외부 서비스
| 서비스 | URL | 용도 |
|--------|-----|------|
| DART OpenAPI | opendart.fss.or.kr | 재무제표 (무료, API 키 필요) |
| KRX | data.krx.co.kr | 주가 데이터 (pykrx 경유) |
| Gitea | git.variet.net | 소스코드 관리 |
| Vikunja | plan.variet.net | 태스크 관리 |
## Python 경로
```
C:\ProgramData\miniforge3\envs\edf\python.exe
```

35
.agents/workflows/check-gitea.md Normal file
View File

@@ -0,0 +1,35 @@
---
description: Gitea API로 저장소 커밋/이슈/PR 현황을 조회하는 워크플로우
---
# Gitea 저장소 현황 조회
서비스 정보는 `.agents/workflows/services.md` 참조.
// turbo-all
## 절차
1. 최근 커밋 조회 (최신 10개):
```powershell
$h = @{Authorization="token 3a01b4b15a39921572e64c413353e870d4d2161b"}
$commits = Invoke-RestMethod -Uri "https://git.variet.net/api/v1/repos/Variet/edf/commits?limit=10&sha=main" -Headers $h
$commits | ForEach-Object { Write-Host "$($_.sha.Substring(0,7)) $($_.commit.message.Split("`n")[0])" }
```
2. 열린 이슈 조회:
```powershell
$h = @{Authorization="token 3a01b4b15a39921572e64c413353e870d4d2161b"}
$issues = Invoke-RestMethod -Uri "https://git.variet.net/api/v1/repos/Variet/edf/issues?state=open&type=issues" -Headers $h
$issues | ForEach-Object { Write-Host "#$($_.number) $($_.title)" }
```
3. Wiki 페이지 목록:
```powershell
C:\ProgramData\miniforge3\envs\edf\python.exe .agents\workflows\helpers\wiki_helper.py list
```
4. Wiki 페이지 읽기:
```powershell
C:\ProgramData\miniforge3\envs\edf\python.exe .agents\workflows\helpers\wiki_helper.py read "Architecture"
```

40
.agents/workflows/check-vikunja.md Normal file
View File

@@ -0,0 +1,40 @@
---
description: Vikunja API로 프로젝트 태스크 현황을 조회하는 워크플로우
---
# Vikunja 태스크 현황 조회
서비스 정보는 `.agents/workflows/services.md` 참조.
// turbo-all
## 절차
1. 전체 목록:
```powershell
C:\ProgramData\miniforge3\envs\edf\python.exe .agents\workflows\helpers\vikunja_helper.py list
```
2. TODO만:
```powershell
C:\ProgramData\miniforge3\envs\edf\python.exe .agents\workflows\helpers\vikunja_helper.py list todo
```
3. DONE만:
```powershell
C:\ProgramData\miniforge3\envs\edf\python.exe .agents\workflows\helpers\vikunja_helper.py list done
```
4. 태스크 완료 처리 (**⚠️ 반드시 이 방법 사용 — 직접 API 호출 금지**):
```powershell
C:\ProgramData\miniforge3\envs\edf\python.exe .agents\workflows\helpers\vikunja_helper.py done {TASK_ID}
```
5. 새 태스크 생성:
```powershell
C:\ProgramData\miniforge3\envs\edf\python.exe .agents\workflows\helpers\vikunja_helper.py create "제목" "설명" --labels Engine,Priority:High
```
> [!CAUTION]
> **절대로** `Invoke-RestMethod -Method Post -Body '{"done": true}'` 같은 직접 API 호출을 사용하지 마세요.
> `vikunja_helper.py`는 항상 GET → 기존 필드 보존 → POST 패턴을 사용합니다.

41
.agents/workflows/debug.md Normal file
View File

@@ -0,0 +1,41 @@
---
description: 에러/버그 발생 시 체계적 디버깅 워크플로우 (에러, 안돼요, 왜 안돼, 버그, 디버그, 수정)
---
# Debug Workflow
> [!IMPORTANT]
> 추측으로 코드를 수정하지 마세요. 반드시 이 순서를 따릅니다.
## 1단계: 정보 수집 (추측 금지)
- [ ] 에러 메시지 **전문** 확인 (절대 잘라내지 않기)
- [ ] 관련 로그 파일 확인
- [ ] 환경 정보 확인 (OS, Python 버전, 의존성 버전 등)
- [ ] 에러가 발생하는 **정확한 입력/조건** 파악
## 2단계: Known Issues 확인
`.agents/references/known-issues.md`를 읽고 동일하거나 유사한 문제가 있는지 확인합니다.
> [!CAUTION]
> **known-issues 확인 없이 해결 시도를 시작하지 마세요.**
## 3단계: 근본 원인 분석
- [ ] 에러가 발생하는 **정확한 코드 위치** 확인
- [ ] 가설을 세우고, 가설을 검증할 수 있는 **최소한의 테스트** 수행
- [ ] 가설이 틀렸다면 **즉시 다른 가설로 전환**
> [!WARNING]
> **동일한 접근을 2회 초과 시도하지 마세요.**
## 4단계: 수정 및 검증
- [ ] 수정 적용
- [ ] 동일 에러가 재현되지 않는지 확인
- [ ] 사이드 이펙트 없는지 확인
## 5단계: 기록
- [ ] `known-issues.md`에 새 항목 추가

108
.agents/workflows/end.md Normal file
View File

@@ -0,0 +1,108 @@
---
description: 세션 종료 시 devlog 기록 + git commit + Vikunja 동기화 (끝, 마무리, 커밋해, 완료)
---
# 세션 종료 프로토콜
작업 완료, "끝", "마무리", "커밋해" 등 요청 시 이 워크플로우를 실행합니다.
// turbo-all
## 0. 학습 기록 (실패/시행착오 저장)
이번 세션에서 발생한 실패, 시행착오, 새로 알게 된 사실을 정리합니다:
- [ ] `.agents/references/known-issues.md`에 추가할 항목이 있는지 확인
- [ ] 있다면 아래 포맷으로 추가:
```markdown
### [날짜] [키워드] — 한줄 요약
- **증상**: ...
- **원인**: ...
- **해결**: ...
- **주의**: ...
```
## 1. Devlog 기록
### Index 업데이트 (필수 — 매 작업)
오늘 날짜의 index 파일에 완료된 작업 1줄을 추가합니다.
- **파일**: `docs/devlog/YYYY-MM-DD.md`
- **형식**:
```markdown
| NNN | HH:MM | 작업 설명 | `커밋해시` | ✅ 또는 🔧 |
```
> [!TIP]
> - ✅ = 완료, 🔧 = 미완료 (다음 세션에서 이어받기)
> - 파일이 없으면 새로 생성 (테이블 헤더 포함)
### Entry 작성 (선택적 — 필요할 때만)
> [!IMPORTANT]
> Entry는 **git/Vikunja/wiki에 없는 정보**가 있을 때만 작성합니다.
**Entry 파일**: `docs/devlog/entries/YYYYMMDD-NNN.md`
---
## 2. Vikunja 동기화
> [!CAUTION]
> **반드시 `vikunja_helper.py` 사용.** 직접 API 호출 금지.
### 2-1. 커밋 전수 검사
이번 세션의 **모든 커밋을 하나씩 검사**하고 Vikunja에 매핑합니다.
```powershell
git log --oneline -20
```
| 커밋 유형 | Vikunja 액션 |
|-----------|-------------|
| 기존 태스크 해당 작업 **완료** | `C:\ProgramData\miniforge3\envs\edf\python.exe .agents\workflows\helpers\vikunja_helper.py done {ID}` |
| 신규 작업 완료 (기존 태스크 없음) | `C:\ProgramData\miniforge3\envs\edf\python.exe .agents\workflows\helpers\vikunja_helper.py create "제목" "설명" --done --labels Engine,Priority:High` |
| 작업 중 발견된 **미완료 TODO** | `C:\ProgramData\miniforge3\envs\edf\python.exe .agents\workflows\helpers\vikunja_helper.py create "제목" "설명" --labels Engine,Priority:Mid` |
### 2-2. 완료 처리
```powershell
C:\ProgramData\miniforge3\envs\edf\python.exe .agents\workflows\helpers\vikunja_helper.py done {TASK_ID}
```
### 2-3. 신규 태스크 생성
```powershell
C:\ProgramData\miniforge3\envs\edf\python.exe .agents\workflows\helpers\vikunja_helper.py create "제목" "설명" --labels Engine,Priority:High
```
---
## 3. Git Commit & Push
```powershell
git add -A
git status --short
```
```powershell
git commit -m "커밋 메시지"
```
```powershell
git push origin main
```
---
## 4. 최종 체크리스트
> [!WARNING]
> 아래 항목 중 하나라도 누락되면 세션 종료를 완료할 수 없습니다.
- [ ] known-issues 업데이트됨 (새 이슈가 있었다면)
- [ ] devlog index 업데이트됨
- [ ] Vikunja 태스크 생성/완료 처리됨 (커밋 전수 검사 기반)
- [ ] git push 완료
- [ ] 사용자에게 완료 보고

216
.agents/workflows/helpers/vikunja_helper.py Normal file
View File

@@ -0,0 +1,216 @@
"""Vikunja safe task updater — preserves existing fields when updating tasks.
Usage:
python vikunja_helper.py done 75 # Mark task #75 as done
python vikunja_helper.py done 71 77 78 # Mark multiple tasks done
python vikunja_helper.py undone 75 # Mark task #75 as not done
python vikunja_helper.py comment 75 "text" # Add comment to task #75
python vikunja_helper.py desc 75 "text" # Set description (appends if exists)
python vikunja_helper.py create "title" "desc" --labels Engine,Priority:High
python vikunja_helper.py create "title" "desc" --done --labels Engine,Priority:Mid
python vikunja_helper.py label 75 Engine Priority:High # Add labels to task
python vikunja_helper.py list # List all tasks
python vikunja_helper.py list todo # List TODO only
python vikunja_helper.py list done # List DONE only
"""
import sys
import json
import urllib.request
import urllib.error
import io
# Fix Windows console encoding (cp949 → utf-8)
if sys.stdout.encoding != "utf-8":
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
# ============================================================
# ⚙️ CONFIGURATION — PROJECT_ID만 프로젝트별로 변경하세요
# ============================================================
API_BASE = "https://plan.variet.net/api/v1"
TOKEN = "tk_070f8e0b715e818bb7178c3815ed5389040eddca"
PROJECT_ID = 11 # EDF project
# ============================================================
HEADERS = {
"Authorization": f"Bearer {TOKEN}",
"Content-Type": "application/json",
}
# Label name → Vikunja label ID mapping
LABEL_MAP = {
"Backend": 1, "Frontend": 2, "Engine": 3, "Infra": 4, "Test": 5,
"Priority:High": 6, "Priority:Mid": 7, "Priority:Low": 8,
"Agent": 17, "Tool": 18, "AI/LLM": 19,
}
def api_get(path: str):
req = urllib.request.Request(f"{API_BASE}{path}", headers=HEADERS)
with urllib.request.urlopen(req) as resp:
return json.loads(resp.read().decode("utf-8"))
def api_post(path: str, data: dict):
body = json.dumps(data).encode("utf-8")
req = urllib.request.Request(f"{API_BASE}{path}", data=body, headers=HEADERS, method="POST")
with urllib.request.urlopen(req) as resp:
return json.loads(resp.read().decode("utf-8"))
def api_put(path: str, data: dict):
body = json.dumps(data).encode("utf-8")
req = urllib.request.Request(f"{API_BASE}{path}", data=body, headers=HEADERS, method="PUT")
with urllib.request.urlopen(req) as resp:
return json.loads(resp.read().decode("utf-8"))
def get_task(task_id: int) -> dict:
return api_get(f"/tasks/{task_id}")
def safe_update_task(task_id: int, updates: dict) -> dict:
task = get_task(task_id)
safe_body = {
"title": task.get("title", ""),
"description": task.get("description", ""),
"priority": task.get("priority", 0),
"done": task.get("done", False),
}
safe_body.update(updates)
return api_post(f"/tasks/{task_id}", safe_body)
def mark_done(task_ids: list):
for tid in task_ids:
result = safe_update_task(tid, {"done": True})
title = result.get("title", "?")
print(f" ✅ #{tid} → done=True [{title}]")
def mark_undone(task_ids: list):
for tid in task_ids:
result = safe_update_task(tid, {"done": False})
title = result.get("title", "?")
print(f" ⬜ #{tid} → done=False [{title}]")
def add_comment(task_id: int, comment: str):
result = api_put(f"/tasks/{task_id}/comments", {"comment": comment})
print(f" 💬 #{task_id} comment added (id={result.get('id', '?')})")
def set_description(task_id: int, desc: str, append: bool = True):
task = get_task(task_id)
existing = task.get("description", "") or ""
if append and existing:
new_desc = existing.rstrip() + "\n\n" + desc
else:
new_desc = desc
result = safe_update_task(task_id, {"description": new_desc})
print(f" 📝 #{task_id} description updated [{result.get('title', '?')}]")
def list_tasks(filter_: str = "all"):
all_tasks = []
page = 1
while True:
batch = api_get(f"/projects/{PROJECT_ID}/tasks?per_page=50&page={page}")
if not batch:
break
all_tasks.extend(batch)
if len(batch) < 50:
break
page += 1
if filter_ == "todo":
all_tasks = [t for t in all_tasks if not t["done"]]
elif filter_ == "done":
all_tasks = [t for t in all_tasks if t["done"]]
all_tasks.sort(key=lambda t: t["id"])
for t in all_tasks:
status = "✅" if t["done"] else "⬜"
desc = (t.get("description") or "")[:50].replace("\n", " ")
labels = ", ".join(l["title"] for l in (t.get("labels") or []))
print(f" {status} #{t['id']:3d} {t['title'][:40]:<40} [{labels}] {desc}")
print(f"\n Total: {len(all_tasks)} tasks")
def add_labels(task_id: int, label_names: list):
for name in label_names:
label_id = LABEL_MAP.get(name)
if not label_id:
print(f" ⚠️ Unknown label '{name}'. Valid: {', '.join(LABEL_MAP.keys())}")
continue
try:
api_put(f"/tasks/{task_id}/labels", {"label_id": label_id})
print(f" 🏷️ #{task_id} + {name} (id={label_id})")
except Exception as e:
if "already" in str(e).lower() or "409" in str(e):
print(f" 🏷️ #{task_id} already has {name}")
else:
print(f" ⚠️ #{task_id} label {name} failed: {e}")
def create_task(title: str, description: str = "", done: bool = False, labels: list = None):
payload = {"title": title, "description": description}
result = api_put(f"/projects/{PROJECT_ID}/tasks", payload)
task_id = result["id"]
print(f" ✨ #{task_id} created: {result.get('title', '?')}")
if labels:
add_labels(task_id, labels)
if done:
result = safe_update_task(task_id, {"done": True})
print(f" ✅ #{task_id} → done=True")
return result
def main():
if len(sys.argv) < 2:
print(__doc__)
return
cmd = sys.argv[1].lower()
if cmd == "done":
ids = [int(x) for x in sys.argv[2:]]
mark_done(ids)
elif cmd == "undone":
ids = [int(x) for x in sys.argv[2:]]
mark_undone(ids)
elif cmd == "comment":
add_comment(int(sys.argv[2]), sys.argv[3])
elif cmd == "desc":
set_description(int(sys.argv[2]), sys.argv[3])
elif cmd == "list":
f = sys.argv[2] if len(sys.argv) > 2 else "all"
list_tasks(f)
elif cmd == "label":
if len(sys.argv) < 4:
print("Usage: vikunja_helper.py label TASK_ID Label1 Label2 ...")
return
add_labels(int(sys.argv[2]), sys.argv[3:])
elif cmd == "create":
title = sys.argv[2] if len(sys.argv) > 2 else ""
desc = sys.argv[3] if len(sys.argv) > 3 and not sys.argv[3].startswith("--") else ""
is_done = "--done" in sys.argv
labels = None
for i, arg in enumerate(sys.argv):
if arg == "--labels" and i + 1 < len(sys.argv):
labels = sys.argv[i + 1].split(",")
break
if not title:
print("Error: title is required")
return
create_task(title, desc, done=is_done, labels=labels)
else:
print(f"Unknown command: {cmd}")
print(__doc__)
if __name__ == "__main__":
main()

100
.agents/workflows/helpers/wiki_helper.py Normal file
View File

@@ -0,0 +1,100 @@
"""Gitea Wiki helper: list, read, create, update wiki pages.
Usage:
wiki_helper.py list — list all pages
wiki_helper.py read <title> — read a page
wiki_helper.py create <title> <file> — create a page from file
wiki_helper.py update <title> <file> — update a page from file
"""
import sys, io, json, base64, urllib.request, urllib.error
sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
# ============================================================
# ⚙️ CONFIGURATION — GITEA_REPO만 프로젝트별로 변경하세요
# ============================================================
GITEA_BASE_URL = "https://git.variet.net"
GITEA_OWNER = "Variet"
GITEA_REPO = "edf" # ← EDF project
GITEA_TOKEN = "3a01b4b15a39921572e64c413353e870d4d2161b"
# ============================================================
BASE = f"{GITEA_BASE_URL}/api/v1/repos/{GITEA_OWNER}/{GITEA_REPO}/wiki"
HEADERS = {"Authorization": f"token {GITEA_TOKEN}", "Content-Type": "application/json"}
def _req(method, path, data=None):
url = f"{BASE}{path}"
body = json.dumps(data).encode() if data else None
req = urllib.request.Request(url, data=body, headers=HEADERS, method=method)
try:
with urllib.request.urlopen(req) as resp:
return json.loads(resp.read().decode())
except urllib.error.HTTPError as e:
err = e.read().decode()
print(f" ⚠️ HTTP {e.code}: {err}")
return None
def _find_sub_url(title):
pages = _req("GET", "/pages")
if pages:
for p in pages:
if p.get("title", "").lower() == title.lower():
return p.get("sub_url", title)
return title
def list_pages():
pages = _req("GET", "/pages")
if pages:
print(f"=== {len(pages)} Wiki Pages ===")
for p in pages:
print(f" {p.get('title', '?')}")
return pages
def read_page(title):
sub = _find_sub_url(title)
page = _req("GET", f"/page/{sub}")
if page and page.get("content_base64"):
content = base64.b64decode(page["content_base64"]).decode("utf-8")
return content
return None
def create_page(title, content):
data = {
"title": title,
"content_base64": base64.b64encode(content.encode()).decode(),
}
result = _req("POST", "/new", data)
if result:
print(f" ✅ Created wiki page: {title}")
return result
def update_page(title, content):
sub = _find_sub_url(title)
data = {
"title": title,
"content_base64": base64.b64encode(content.encode()).decode(),
}
result = _req("PATCH", f"/page/{sub}", data)
if result:
print(f" ✅ Updated wiki page: {title}")
return result
if __name__ == "__main__":
cmd = sys.argv[1] if len(sys.argv) > 1 else "list"
if cmd == "list":
list_pages()
elif cmd == "read" and len(sys.argv) > 2:
content = read_page(sys.argv[2])
if content:
print(content[:5000])
else:
print(f" Page '{sys.argv[2]}' not found")
elif cmd == "create" and len(sys.argv) > 3:
with open(sys.argv[3], "r", encoding="utf-8") as f:
create_page(sys.argv[2], f.read())
elif cmd == "update" and len(sys.argv) > 3:
with open(sys.argv[3], "r", encoding="utf-8") as f:
update_page(sys.argv[2], f.read())
else:
print("Usage: wiki_helper.py list|read <title>|create <title> <file>|update <title> <file>")

39
.agents/workflows/pre-task.md Normal file
View File

@@ -0,0 +1,39 @@
---
description: 모든 구현 작업 전 실행하는 사전 점검 체크리스트 (pre-task, 준비, 시작 전, 계획, 구현)
---
# Pre-Task Checklist
> [!IMPORTANT]
> 코딩을 시작하기 전에 반드시 이 체크리스트를 순서대로 완료하세요.
> 체크리스트를 건너뛸 경우 불필요한 시행착오가 발생합니다.
## 1단계: 요구사항 정리
- [ ] 유저 요청을 구체적 작업 항목으로 분해
- [ ] 변경 범위(scope)를 명확히 정의 (영향받는 파일/모듈)
- [ ] 성공 기준(acceptance criteria) 확인
## 2단계: 레퍼런스 확인 (추측 금지)
- [ ] `.agents/references/architecture.md` — 현재 아키텍처 확인
- [ ] `.agents/references/tech-stack.md` — 기술 스택 및 버전 확인
- [ ] `.agents/references/conventions.md` — 코딩 컨벤션 확인
- [ ] `.agents/references/known-issues.md` — 과거 실패 패턴 확인
- [ ] 관련 기존 코드 최소 3개 파일 읽기
> [!CAUTION]
> 레퍼런스 문서가 존재하는 주제에 대해 추측하지 마세요.
> 문서가 없으면 유저에게 확인을 요청하세요.
## 3단계: 계획 수립
- [ ] 변경할 파일 목록 작성
- [ ] 의존성 순서 파악 (어떤 파일부터 수정해야 하는가?)
- [ ] 리스크 식별 (어디서 실패할 가능성이 높은가?)
- [ ] 테스트 방법 결정 (어떻게 검증할 것인가?)
## 4단계: 유저 확인
- [ ] 계획을 유저에게 보고하고 승인받기 (변경 파일 3개 이상인 경우)
- [ ] 작은 변경은 바로 실행하되, 변경 내용을 명확히 설명

106
.agents/workflows/services.md Normal file
View File

@@ -0,0 +1,106 @@
---
description: 프로젝트 연동 서비스 URL, API 키, 프로젝트 정보 참조
---
# 서비스 연동 정보
> [!CAUTION]
> 이 파일에는 API 토큰이 포함되어 있습니다. 프라이빗 레포에서만 사용하세요.
## 로컬 환경
| 항목 | 값 |
|------|-----|
| **Python** | `C:\ProgramData\miniforge3\envs\edf\python.exe` (**항상 이 경로 사용**) |
| **Shell** | PowerShell (`curl` = `Invoke-WebRequest` 별칭이므로 반드시 **`curl.exe`** 사용) |
## Gitea (Git Repository)
| 항목 | 값 |
|------|-----|
| **Base URL** | `https://git.variet.net` |
| **API Base** | `https://git.variet.net/api/v1` |
| **Repo** | `Variet/edf` |
| **Token** | `3a01b4b15a39921572e64c413353e870d4d2161b` |
| **Auth Header** | `-H "Authorization: token 3a01b4b15a39921572e64c413353e870d4d2161b"` |
## Vikunja (Task Management)
| 항목 | 값 |
|------|-----|
| **Base URL** | `https://plan.variet.net` |
| **API Base** | `https://plan.variet.net/api/v1` |
| **Project ID** | `11` |
| **Token** | `tk_070f8e0b715e818bb7178c3815ed5389040eddca` |
| **Auth Header** | `-H "Authorization: Bearer tk_070f8e0b715e818bb7178c3815ed5389040eddca"` |
## Vikunja 태스크 조회
> [!TIP]
> 태스크 목록은 항상 라이브 조회를 사용합니다. 하드코딩된 매핑은 유지하지 않습니다.
```powershell
C:\ProgramData\miniforge3\envs\edf\python.exe .agents\workflows\helpers\vikunja_helper.py list todo
```
## AI 작업 프로토콜
> [!IMPORTANT]
> 아래 규칙은 모든 작업에 자동 적용됩니다. 유저가 별도 지시하지 않아도 따릅니다.
### Vikunja = Single Source of Truth (SSOT)
- **Vikunja가 유일한 작업 현황 관리 도구**입니다.
- 로컬 `task.md`는 현재 대화 내 세부 체크리스트용으로만 사용합니다.
- 새 TODO 발견 시 → Vikunja에 태스크 생성 (로컬 파일에만 적는 것은 금지)
- 작업 완료 시 → Vikunja 태스크 완료 처리 (로컬 체크만 하는 것은 금지)
### Vikunja 태깅 규칙
태스크 생성 시 반드시 아래 라벨을 적절히 부여합니다:
**영역 라벨 (필수, 1개 이상):**
| ID | 라벨 | 적용 대상 |
|:--:|-------|-----------:|
| 1 | `Backend` | 서버, DB, API |
| 2 | `Frontend` | UI, 웹 프론트엔드 |
| 3 | `Engine` | 핵심 엔진/로직 |
| 4 | `Infra` | Docker, CI/CD, 모니터링 |
| 5 | `Test` | 테스트, E2E |
**우선순위 라벨 (필수, 1개):**
| ID | 라벨 | 기준 |
|:--:|-------|------:|
| 6 | `Priority:High` | 핵심 기능 미완성, 블로커 |
| 7 | `Priority:Mid` | 기능 개선, UX 향상, 리팩터링 |
| 8 | `Priority:Low` | nice-to-have, 문서, 코드 정리 |
### 작업 시작 시
1. `git pull` 으로 최신 코드 동기화
2. Vikunja 태스크 조회 (`/check-vikunja`) → 관련 태스크 ID 확인
3. 관련 태스크가 있으면 Vikunja에서 진행중 표시
4. 관련 태스크가 없으면 Vikunja에 새 태스크 생성 (태깅 규칙 준수)
### 작업 중
5. 의미 있는 단위마다 자주 커밋 (대규모 변경을 한번에 커밋하지 않음)
6. 커밋 메시지 규칙:
- `feat:`, `fix:`, `refactor:`, `test:`, `docs:`, `chore:`, `ci:` 접두사 사용
- 관련 Vikunja 태스크가 있으면 `#task-{ID}` 참조 포함
### 작업 완료 시
7. 모든 변경사항 커밋 + `git push`
8. Vikunja 태스크 완료 처리 (**반드시 `vikunja_helper.py` 사용**):
```powershell
C:\ProgramData\miniforge3\envs\edf\python.exe .agents\workflows\helpers\vikunja_helper.py done {TASK_ID}
```
> [!CAUTION]
> **직접 `Invoke-RestMethod -Body '{"done": true}'` 사용 금지!**
> Vikunja API는 POST 시 body에 없는 필드를 빈값으로 덮어씁니다.
## PowerShell 주의사항
- `curl` → PowerShell에서 `Invoke-WebRequest`의 별칭. **반드시 `curl.exe`** 사용
- JSON 파이프 파싱 시 PowerShell 이스케이핑 문제 → `.py` 스크립트 파일로 만들어 실행 권장

66
.agents/workflows/start.md Normal file
View File

@@ -0,0 +1,66 @@
---
description: 세션 시작 시 프로젝트 맥락을 빠르게 복구하는 워크플로우 (시작, continue, 이어서, 작업 시작)
---
# 세션 시작 프로토콜
새 대화 시작, "continue", "이어서", "작업 시작" 등 요청 시 이 워크플로우를 실행합니다.
// turbo-all
## 절차
### 0. 에이전트 룰 & 맥락 로딩 (자동)
`.agents/AGENT.md`를 읽고 에이전트 행동 규칙을 로딩합니다.
`.agents/references/known-issues.md`를 읽어 최근 이슈를 파악합니다.
`.agents/workflows/services.md`의 **로컬 환경** 섹션을 읽고 Python 경로 등 환경 설정을 확인합니다.
### 1. Devlog 맥락 복구
오늘 + 어제 devlog index를 읽고 최근 작업 흐름을 파악합니다.
```powershell
$today = Get-Date -Format "yyyy-MM-dd"
$yesterday = (Get-Date).AddDays(-1).ToString("yyyy-MM-dd")
if (Test-Path "docs\devlog\$today.md") {
Write-Host "=== Devlog: $today ==="
Get-Content "docs\devlog\$today.md"
} elseif (Test-Path "docs\devlog\$yesterday.md") {
Write-Host "=== Devlog: $yesterday (no entry for today yet) ==="
Get-Content "docs\devlog\$yesterday.md"
} else {
Write-Host "=== No recent devlog found ==="
}
```
미완료(🔧) 항목이 있으면 해당 entry 파일을 읽어 이어받기 맥락을 확보합니다:
- Entry 경로: `docs/devlog/entries/YYYYMMDD-NNN.md`
### 2. Git 상태 확인
```powershell
git status --short
```
```powershell
git log --oneline -5
```
### 3. Vikunja TODO 태스크
```powershell
C:\ProgramData\miniforge3\envs\edf\python.exe .agents\workflows\helpers\vikunja_helper.py list todo
```
### 4. 종합 보고
결과를 종합하여 사용자에게 보고:
- 마지막 작업 맥락 + 미완료 항목 (devlog 🔧 기반)
- TODO 태스크 목록 (라벨 + 우선순위)
- 다음 작업 제안
**우선순위 판단 기준** (라벨만으로 판단 금지):
- P0: 최근 커밋에서 스키마/모델/인터페이스 변경 → 연쇄 영향 점검
- P1: 서버 기동/API 응답 장애
- P2: 기능 미완성/UX 개선
- P3: 정확도 향상, 신규 기능, CI/CD, 문서 정리

26
.gitignore vendored Normal file
View File

@@ -0,0 +1,26 @@
# Python
__pycache__/
*.py[cod]
*.egg-info/
dist/
build/
*.egg
# Data files (large, don't commit to git)
data/
outputs/
docs_cache/
*.db
*.pkl
# Environment
.env
*.log
# IDE
.vscode/
.idea/
# OS
Thumbs.db
.DS_Store

27
README.md Normal file
View File

@@ -0,0 +1,27 @@
# EDF Project: 주식 변동성 기반 한국 등급별 부도율 산출
## Overview
KRX 상장 한국 기업의 주가 변동성을 활용하여 Merton-KMV 모형 기반 신용등급별 부도율을 산출하는 프로젝트
## Quick Start
```bash
pip install -r requirements.txt
# config/settings.yaml에 DART API 키 설정 확인
python -m src.data.krx_fetcher # 주가 데이터 수집
python -m src.data.dart_fetcher # 재무제표 수집
python -m src.models.merton # DD/EDF 산출
```
## Project Structure
```
EDF/
├── config/settings.yaml # API 키, 파라미터 설정
├── src/
│ ├── data/ # 데이터 수집 모듈
│ ├── models/ # Merton, DD, Shadow Rating
│ ├── calibration/ # 보정, 블렌딩
│ └── validation/ # 백테스팅, 검증
├── data/ # 수집된 데이터 저장
├── outputs/ # 결과물
└── docs/ # 기술 문서
```

39
config/settings.yaml Normal file
View File

@@ -0,0 +1,39 @@
# DART API
dart_api_key: "ef6deb100be436aed88051fd4914dbdb58ff2e94"
# 분석 기간
start_year: 2019
end_year: 2025
# Merton 모형 파라미터
merton:
time_horizon: 1.0 # T (년)
risk_free_rate: 0.035 # 무위험이자율 (기본값, ECOS에서 업데이트 가능)
volatility_window: 252 # 변동성 추정 윈도우 (거래일)
volatility_method: "historical" # historical / ewma / garch
ewma_lambda: 0.94 # EWMA λ
default_point_ltd_ratio: 0.5 # 장기부채 가중치
# Shadow Rating
shadow_rating:
model_type: "ordered_probit" # ordered_probit / ordered_logit
min_samples_per_grade: 30 # 등급별 최소 표본수
# 글로벌 블렌딩
blending:
threshold: 50 # 표본수 임계치
bayesian_prior_strength: 50 # 베이지안 사전 강도
# 데이터 경로
paths:
raw_data: "data/raw"
processed_data: "data/processed"
external_data: "data/external"
outputs: "outputs"
# KRX 설정
krx:
markets: ["KOSPI", "KOSDAQ"]
exclude_sectors: ["금융업", "보험업"] # 금융업 제외
min_trading_days: 200 # 최소 거래일수 필터
sleep_seconds: 0.5 # API 호출 간 대기

827
docs/technical_methodology.md Normal file
View File

@@ -0,0 +1,827 @@
# 주식 변동성 기반 등급별 부도율 산출 — 기술 문서
> **프로젝트**: KRX 상장 한국 기업 대상 Equity Volatility → Default Rate by Rating
> **작성일**: 2026-03-11
> **버전**: v0.1 (초안)
---
## 목차
1. [이론적 기초](#1-이론적-기초)
2. [핵심 수학적 프레임워크](#2-핵심-수학적-프레임워크)
3. [한국 시장 등급 관측 문제 및 대안](#3-한국-시장-등급-관측-문제-및-대안)
4. [글로벌 접근 방법론 비교](#4-글로벌-접근-방법론-비교)
5. [구현 아키텍처](#5-구현-아키텍처)
6. [데이터 명세](#6-데이터-명세)
7. [알고리즘 상세](#7-알고리즘-상세)
8. [검증 방법론](#8-검증-방법론)
9. [한국 시장 특수 고려사항](#9-한국-시장-특수-고려사항)
10. [기술 스택 및 의존성](#10-기술-스택-및-의존성)
11. [참고 문헌](#11-참고-문헌)
---
## 1. 이론적 기초
### 1.1 구조적 모형(Structural Model) 계보
```
Black-Scholes (1973)
└─ Merton (1974) ─── 기업부도를 옵션으로 해석
├─ Black-Cox (1976) ─── First Passage Time (배리어 부도)
├─ Geske (1977) ─── 복합옵션 (쿠폰부 부채)
├─ Longstaff-Schwartz (1995) ─── 확률적 이자율
└─ KMV (Kealhofer-McQuown-Vasicek)
└─ Moody's Analytics EDF™ ─── 상용화
```
### 1.2 Merton 모형 (1974)
**핵심 가정:**
- 기업의 자산가치 `V(t)`는 기하 브라운 운동(GBM)을 따름
- 부채는 만기 `T`에 원금 `D`가 일시 상환되는 제로쿠폰 채권
- 자기자본 `E`는 자산 `V`에 대한 유럽형 콜옵션
**자산 역학:**
```
dV = μ·V·dt + σ_V·V·dW
```
- `μ`: 자산 기대수익률 (drift)
- `σ_V`: 자산 변동성
- `W`: 위너 과정
**부도 조건:**
```
Default ⟺ V(T) < D (만기 시점에 자산가치 < 부채)
```
**자기자본의 옵션 해석:**
```
E = Call(V, D, T) = V·N(d₁) - D·e^{-rT}·N(d₂)
```
### 1.3 KMV-Moody's EDF 모형
Merton 모형의 실무 확장:
| 구분 | Merton 원형 | KMV 수정 |
|------|-------------|----------|
| 부도점 | D (총부채) | STD + 0.5×LTD |
| 부도 시점 | 만기 T 시점만 | 임의 시점 (First Passage) |
| EDF 산출 | N(-DD) 이론값 | 경험적 부도 빈도 매핑 |
| 데이터 | 단일 시점 | 시계열 반복 추정 |
### 1.4 축약형 모형(Reduced-Form Model)
**CreditRisk+ (Credit Suisse)**
- 부도를 포아송 과정으로 모형화
- 부도율의 변동성을 명시적 반영 (부도율 자체가 확률변수)
- 섹터별 체계적 요인으로 부도 상관관계 간접 포착
- 장점: 구현 용이, 대규모 포트폴리오 적합
- 한계: 시장 데이터 반영 제한, 등급전이 미반영
**Jarrow-Turnbull / Duffie-Singleton**
- 부도 강도(hazard rate)가 시장 변수에 의존
- CDS/채권 스프레드에서 내재 부도확률 추출
- 한국 적용 한계: CDS 시장 유동성 부족
### 1.5 CreditMetrics 접근법
- 등급전이행렬(Rating Transition Matrix) 기반
- 잠재 변수 `Zt`를 통해 체계적 리스크 반영
- 전이확률 × 등급별 스프레드 → 포트폴리오 가치 분포
- 한국 시장: 신평사 발표 전이행렬과 연동 가능
---
## 2. 핵심 수학적 프레임워크
### 2.1 Merton 연립방정식
관측 가능한 `(E, σ_E)`로부터 비관측 `(V, σ_V)`를 추정:
**방정식 1 — 자기자본 가치:**
```
E = V·N(d₁) - D·e^{-rT}·N(d₂)
```
**방정식 2 — 변동성 관계 (Itô's Lemma):**
```
σ_E = (V/E)·N(d₁)·σ_V
```
**여기서:**
```
d₁ = [ln(V/D) + (r + σ²_V/2)·T] / (σ_V·√T)
d₂ = d₁ - σ_V·√T
```
### 2.2 Distance-to-Default (DD)
```
DD = [ln(V/DP) + (μ - σ²_V/2)·T] / (σ_V·√T)
```
- `DP = STD + 0.5 × LTD` (KMV 부도점)
- `μ`: 자산 기대수익률 (실무에서는 r 또는 과거 추정값 사용)
**해석:** DD는 자산가치가 부도점까지 하락하는 데 필요한 표준편차 수
### 2.3 EDF 산출
**이론적 EDF (정규분포 가정):**
```
EDF_theoretical = N(-DD) = Φ(-DD)
```
**경험적 EDF (KMV 방식):**
```
EDF_empirical = (DD 구간별 실제 부도 기업 수) / (DD 구간별 전체 기업 수)
```
**한국 시장 보정 EDF:**
```
EDF_KR = EDF_theoretical × Calibration_Factor(rating_grade)
```
### 2.4 주가 변동성 추정 방법
#### (a) 역사적 변동성 (Historical Volatility)
```
σ_E = √(252) × std(ln(P_t / P_{t-1}))
```
- 일별 로그수익률의 표준편차 × √252 (연환산)
- 추정 윈도우: 1년 (약 250 거래일)
#### (b) EWMA (Exponentially Weighted Moving Average)
```
σ²_t = λ·σ²_{t-1} + (1-λ)·r²_{t-1}
```
- `λ = 0.94` (RiskMetrics 표준)
- 최근 변동에 더 높은 가중치
#### (c) GARCH(1,1)
```
σ²_t = ω + α·ε²_{t-1} + β·σ²_{t-1}
```
- `ω, α, β`: 최대우도추정(MLE)으로 산출
- `α + β < 1` (정상성 조건)
- 변동성 클러스터링 반영 가능
---
## 3. 한국 시장 등급 관측 문제 및 대안
### 3.1 문제 진단
```
KRX 상장사 약 2,500개
├── 신용등급 보유: 약 500~600개 (주로 회사채/CP 발행사)
│ ├── 관측 가능 등급: BBB ~ A 중심 (약 70%)
│ ├── 고등급 (AAA~AA): 소수 (우량사, 등급 불필요)
│ └── 저등급 (B 이하): 극소수 (상장 유지 자체가 어려움)
└── 신용등급 미보유: 약 1,900개 (소형주, 미발행사)
```
**등급별 관측 분포 추정:**
| 등급군 | KRX 상장사 중 비율 | 부도 관측 가능성 | 주요 이슈 |
|--------|-------------------|-----------------|-----------|
| AAA~AA | ~5% | 극히 낮음 | 부도 사례 거의 0 |
| A | ~20% | 낮음 | 부도 희소하나 관측 가능 |
| BBB | ~40% | 보통 | 가장 관측 풍부 |
| BB | ~20% | 높음 | 투기등급 진입, 데이터 확보 |
| B 이하 | ~5% | 높으나 표본 부족 | 상장폐지와 혼재 |
| 무등급 | ~10% (등급보유 대비) | Shadow 필요 | 대부분 소형주 |
### 3.2 대안 전략 상세
#### 대안 1: Shadow Rating (내재등급) 모형
**목적:** DD 및 재무비율을 기반으로 무등급 기업에 내재등급 부여
**방법론 — Ordered Probit 모형:**
```
y* = β'X + ε, ε ~ N(0, 1)
y = k if τ_{k-1} < y* ≤ τ_k
여기서:
- y: 관측등급 (AAA=1, AA+=2, ..., D=n)
- X: [DD, log(총자산), 부채비율, 이자보상비율, EBITDA마진, ROA, 유동비율, 산업더미]
- τ_k: 등급 경계 절단점(cutoff)
```
**학습 과정:**
1. 등급 보유 기업의 (X, y) 쌍으로 β, τ를 MLE 추정
2. 등급 미보유 기업에 추정된 β'X를 적용하여 각 등급 확률 계산
3. 최대 확률 등급을 Shadow Rating으로 부여
**설명변수 후보:**
| 변수 | 정의 | 기대 부호 |
|------|------|-----------|
| DD | Distance-to-Default | + (높을수록 고등급) |
| log_assets | ln(총자산) | + 규모 효과 |
| leverage | 총부채/총자산 | - |
| int_coverage | EBITDA/이자비용 | + |
| ebitda_margin | EBITDA/매출 | + |
| roa | 순이익/총자산 | + |
| current_ratio | 유동자산/유동부채 | + |
| cash_ratio | 현금/유동부채 | + |
| industry | 산업 더미 | 산업별 상이 |
#### 대안 2: DD-Rating 직접 매핑
글로벌 벤치마크를 기반으로 DD 구간 → 등급 매핑:
| DD 범위 | Moody's 등급 | 한국 등급 (추정) | 이론적 EDF |
|---------|-------------|-----------------|------------|
| > 6.0 | Aaa ~ Aa1 | AAA ~ AA+ | < 0.02% |
| 5.0 ~ 6.0 | Aa2 ~ Aa3 | AA ~ AA- | 0.02% ~ 0.05% |
| 4.0 ~ 5.0 | A1 ~ A3 | A+ ~ A- | 0.05% ~ 0.20% |
| 3.0 ~ 4.0 | Baa1 ~ Baa2 | BBB+ ~ BBB | 0.20% ~ 0.70% |
| 2.5 ~ 3.0 | Baa3 | BBB- | 0.70% ~ 1.50% |
| 2.0 ~ 2.5 | Ba1 | BB+ | 1.50% ~ 3.00% |
| 1.5 ~ 2.0 | Ba2 ~ Ba3 | BB ~ BB- | 3.00% ~ 5.00% |
| 1.0 ~ 1.5 | B1 ~ B2 | B+ ~ B | 5.00% ~ 10.00% |
| 0.5 ~ 1.0 | B3 ~ Caa1 | B- ~ CCC+ | 10.00% ~ 20.00% |
| < 0.5 | Caa2 이하 | CCC 이하 | > 20.00% |
> 주의: 글로벌 매핑은 한국 시장에 직접 적용 시 보정(calibration) 필수
#### 대안 3: 등급군 병합(Grade Pooling)
표본 부족 등급을 인접 등급과 통합:
```
Pool 1: AAA + AA+ + AA + AA- → "최우량군" (Super-Prime)
Pool 2: A+ + A + A- → "우량군" (Prime)
Pool 3: BBB+ + BBB + BBB- → "투자적격군" (Investment)
Pool 4: BB+ + BB + BB- → "투기등급군" (Speculative)
Pool 5: B+ 이하 → "고위험군" (High-Risk)
```
**병합 기준:**
- 각 풀 내 최소 관측수: 30개 이상 (통계적 유의성)
- Hosmer-Lemeshow 검정 등으로 풀 내 균질성 확인
#### 대안 4: 글로벌 데이터 블렌딩
한국 데이터와 글로벌 벤치마크를 표본수 기반 가중 혼합:
```
DR_blended(g) = w(g) × DR_KR(g) + [1 - w(g)] × DR_Global(g)
w(g) = min(1, N_KR(g) / N_threshold)
```
- `DR_KR(g)`: 한국 등급 g의 관측 부도율
- `DR_Global(g)`: Moody's/S&P 등급 g의 글로벌 부도율
- `N_KR(g)`: 한국 등급 g의 관측 표본수
- `N_threshold`: 신뢰도 임계치 (예: 50)
#### 대안 5: 베이지안 보정
```
사전분포(Prior): π(θ_g) ~ Beta(α_0, β_0) ← 글로벌 부도율에서 유도
우도(Likelihood): L(data|θ_g) = θ_g^d × (1-θ_g)^{n-d}
사후분포(Posterior): π(θ_g|data) ~ Beta(α_0 + d, β_0 + n - d)
여기서:
- θ_g: 등급 g의 실제 부도율 (추정 대상)
- d: 한국 등급 g에서 관측된 부도 건수
- n: 한국 등급 g에서 관측된 전체 기업수
- α_0, β_0: 글로벌 데이터에서 유도된 사전 파라미터
```
**장점:** 표본 부족 등급에서 글로벌 Prior에 자연스럽게 의존, 표본 충분 등급에서는 한국 데이터 위주로 수렴
---
## 4. 글로벌 접근 방법론 비교
| 방법론 | 모형 유형 | 핵심 입력 | 장점 | 한계 | 한국 적용성 |
|--------|-----------|-----------|------|------|-------------|
| **Merton-KMV** | 구조적 | 주가, 부채 | 시장기반, 전향적 | 상장사 한정, 분포가정 | ★★★★★ |
| **CreditMetrics** | 전이행렬 | 등급전이, 스프레드 | 포트폴리오 리스크 | 등급 의존적 | ★★★☆☆ |
| **CreditRisk+** | 축약형 | 부도율, 변동성 | 구현 용이 | 시장 미반영 | ★★☆☆☆ |
| **Jarrow-Turnbull** | 축약형 | CDS스프레드 | 시장가격 반영 | CDS시장 미발달 | ★★☆☆☆ |
| **Altman Z-Score** | 판별분석 | 재무비율 | 간단, 검증됨 | 시장변동 미반영 | ★★★☆☆ |
| **ML (XGBoost)** | 비모수 | 다양한 데이터 | 유연, 비선형 | 해석부족, 과적합 | ★★★☆☆ |
| **Bharath-Shumway** | 구조적(간편) | 주가, 부채 | 단순 구현 | 정밀도 한계 | ★★★★☆ |
### Bharath-Shumway 간편 DD (Naïve DD)
반복 추정 없이 직접 DD 계산 (실무 빠른 적용용):
```
V_naive = E + D
σ_V_naive = (E/(E+D)) × σ_E + (D/(E+D)) × (0.05 + 0.25×σ_E)
DD_naive = [ln(V_naive/D) + (μ - σ²_V_naive/2)×T] / (σ_V_naive × √T)
```
- Bharath & Shumway (2008) 연구에서 반복추정 DD와 유사한 부도예측력 보고
- 대규모 데이터 처리 시 1차 필터로 유용
---
## 5. 구현 아키텍처
### 5.1 시스템 구조
```
EDF/
├── config/
│ ├── settings.yaml # 전역 설정 (기간, 파라미터)
│ └── rating_mapping.yaml # DD-등급 매핑 테이블
├── data/
│ ├── raw/ # 원본 데이터
│ ├── processed/ # 전처리된 데이터
│ └── external/ # 글로벌 부도율 통계
├── src/
│ ├── data/
│ │ ├── krx_fetcher.py # KRX 주가 수집
│ │ ├── dart_fetcher.py # DART 재무제표 수집
│ │ ├── rating_fetcher.py # 신용등급 수집
│ │ └── preprocessor.py # 데이터 전처리
│ ├── models/
│ │ ├── merton.py # Merton 연립방정식 풀이
│ │ ├── dd_calculator.py # DD/EDF 산출
│ │ ├── shadow_rating.py # Shadow Rating 모형
│ │ └── volatility.py # 변동성 추정 (Historical/EWMA/GARCH)
│ ├── calibration/
│ │ ├── global_benchmark.py # 글로벌 벤치마크 로딩
│ │ ├── blending.py # 블렌딩/베이지안 보정
│ │ └── grade_pooling.py # 등급군 병합
│ ├── validation/
│ │ ├── backtesting.py # 백테스팅
│ │ ├── discriminatory.py # 변별력 검증 (ROC, KS, CAP)
│ │ └── calibration_test.py # 보정력 검증 (Hosmer-Lemeshow)
│ └── utils/
│ ├── financial.py # 재무비율 계산
│ └── statistics.py # 통계 유틸리티
├── notebooks/
│ ├── 01_data_exploration.ipynb
│ ├── 02_merton_analysis.ipynb
│ ├── 03_shadow_rating.ipynb
│ └── 04_default_rate_output.ipynb
├── outputs/
│ ├── dd_results/ # DD/EDF 산출 결과
│ ├── rating_results/ # 등급별 부도율 결과
│ └── reports/ # 검증 보고서
├── docs/
│ └── technical_methodology.md # 이 문서
├── requirements.txt
└── README.md
```
### 5.2 처리 파이프라인
```
[Phase 1: 데이터 수집]
KRX 주가 → 전처리 → 주가변동성 산출
DART 재무제표 → 부채 구조(STD/LTD) 추출
신평사 등급 → 연도별 등급 스냅샷
[Phase 2: Merton-KMV 모형]
(E, σ_E, D, r, T) → 반복 추정 → (V, σ_V) → DP → DD → EDF
[Phase 3: Shadow Rating]
등급 보유 기업: (DD, 재무비율) ↔ 등급 매핑 학습
등급 미보유 기업: 학습 모형 → Shadow Rating 부여
[Phase 4: 등급별 부도율 집계]
실제등급 + Shadow Rating → 등급별 연간 부도율 집계
표본 부족 등급 → 글로벌 블렌딩 / 등급군 병합
[Phase 5: 검증]
백테스팅, 변별력/보정력 검증, CRA 발표 데이터 비교
```
---
## 6. 데이터 명세
### 6.1 필수 데이터
| 데이터 항목 | 소스 | 수집 주기 | 필수/선택 |
|-------------|------|-----------|-----------|
| 일별 종가 | KRX / pykrx | 일별 | 필수 |
| 시가총액 | KRX / pykrx | 일별 | 필수 |
| 발행주식수 | KRX / DART | 분기 | 필수 |
| 유동부채 (STD) | DART 재무제표 | 분기/연간 | 필수 |
| 비유동부채 (LTD) | DART 재무제표 | 분기/연간 | 필수 |
| 총자산 | DART 재무제표 | 분기/연간 | 필수 |
| 신용등급 | 한기평/한신평/나이스 | 연간 스냅샷 | 필수 |
| 무위험이자율 | 한국은행(ECOS) | 일별 | 필수 |
| 부도/워크아웃 이력 | KRX 상장폐지, 뉴스 | 사건 기반 | 필수 |
### 6.2 보조 데이터 (Shadow Rating 강화용)
| 데이터 항목 | 소스 | 용도 |
|-------------|------|------|
| EBITDA | DART | 이자보상비율, 마진 |
| 이자비용 | DART | 이자보상비율 |
| 매출액 | DART | EBITDA 마진 |
| 현금 및 현금성 자산 | DART | 유동성 비율 |
| 산업분류코드 (KSIC) | KRX / DART | 산업 더미 변수 |
| 거래량 | KRX | 유동성 필터링 |
### 6.3 글로벌 벤치마크 데이터
| 데이터 | 소스 | 내용 |
|--------|------|------|
| 등급별 연간 부도율 | Moody's Annual Default Study | Aaa~C 20년+ 평균 |
| 등급별 누적 부도율 | S&P Global Default Study | AAA~D 1~20년 |
| 한국 등급별 부도율 | 한기평/한신평/나이스 연간 발표 | 국내 기준 |
### 6.4 부도(Default) 정의
```
다음 이벤트 중 하나 이상 발생 시 "부도"로 정의:
1. 법정관리(회생절차) 개시 결정
2. 워크아웃(채권단 자율협약) 개시
3. 상장폐지 (재무 사유: 자본잠식, 감사의견 거절 등)
4. 부도어음/부도수표 발생
5. 기업회생절차 신청
6. 파산 선고
※ 제외: 합병·분할·자진 상장폐지 등 비재무적 사유
```
---
## 7. 알고리즘 상세
### 7.1 Merton 연립방정식 풀이
```python
import numpy as np
from scipy.optimize import fsolve
from scipy.stats import norm
def solve_merton(E: float, sigma_E: float, D: float,
r: float, T: float = 1.0) -> tuple[float, float]:
"""
Merton 연립방정식을 풀어 자산가치(V)와 자산변동성(σ_V)을 추정.
Parameters
----------
E : float
자기자본 시장가치 (시가총액, 억원)
sigma_E : float
주가수익률 변동성 (연환산, 예: 0.30 = 30%)
D : float
부도점 = STD + 0.5 × LTD (억원)
r : float
무위험이자율 (연, 예: 0.035 = 3.5%)
T : float
시간 수평선 (년, 기본 1.0)
Returns
-------
V : float
추정 자산가치 (억원)
sigma_V : float
추정 자산변동성 (연환산)
"""
def equations(params):
V, sigma_V = params
d1 = (np.log(V / D) + (r + 0.5 * sigma_V**2) * T) / (sigma_V * np.sqrt(T))
d2 = d1 - sigma_V * np.sqrt(T)
eq1 = V * norm.cdf(d1) - D * np.exp(-r * T) * norm.cdf(d2) - E
eq2 = (V / E) * norm.cdf(d1) * sigma_V - sigma_E
return [eq1, eq2]
# 초기값: V0 = E + D, sigma_V0 = sigma_E * E / (E + D)
V0 = E + D
sigma_V0 = sigma_E * E / (E + D)
solution = fsolve(equations, [V0, sigma_V0], full_output=True)
V, sigma_V = solution[0]
return max(V, E), max(sigma_V, 0.01) # 하한 설정
def calculate_dd(V: float, sigma_V: float, D: float,
mu: float, T: float = 1.0) -> float:
"""Distance-to-Default 산출"""
if D <= 0 or V <= 0 or sigma_V <= 0:
return np.nan
DD = (np.log(V / D) + (mu - 0.5 * sigma_V**2) * T) / (sigma_V * np.sqrt(T))
return DD
def calculate_edf(DD: float) -> float:
"""이론적 EDF 산출 (정규분포 가정)"""
if np.isnan(DD):
return np.nan
return norm.cdf(-DD)
```
### 7.2 변동성 추정
```python
def historical_volatility(prices: np.ndarray, window: int = 252) -> float:
"""역사적 변동성 (연환산)"""
log_returns = np.diff(np.log(prices))
if len(log_returns) < window:
window = len(log_returns)
return np.std(log_returns[-window:]) * np.sqrt(252)
def ewma_volatility(prices: np.ndarray, lmbda: float = 0.94) -> float:
"""EWMA 변동성 (연환산)"""
log_returns = np.diff(np.log(prices))
variance = log_returns[0]**2
for ret in log_returns[1:]:
variance = lmbda * variance + (1 - lmbda) * ret**2
return np.sqrt(variance * 252)
def garch_volatility(prices: np.ndarray) -> float:
"""GARCH(1,1) 변동성 (arch 패키지 사용)"""
from arch import arch_model
log_returns = np.diff(np.log(prices)) * 100 # 백분율
model = arch_model(log_returns, vol='Garch', p=1, q=1, dist='normal')
result = model.fit(disp='off')
# 최신 조건부 변동성을 연환산
cond_vol = result.conditional_volatility[-1] / 100
return cond_vol * np.sqrt(252)
```
### 7.3 Shadow Rating (Ordered Probit)
```python
import statsmodels.api as sm
import pandas as pd
def fit_shadow_rating_model(df_rated: pd.DataFrame,
feature_cols: list,
rating_col: str = 'rating_numeric') -> object:
"""
등급 보유 기업 데이터로 Ordered Probit 모형 적합.
Parameters
----------
df_rated : pd.DataFrame
등급 보유 기업 데이터 (DD, 재무비율, 등급 포함)
feature_cols : list
설명변수 컬럼명 리스트
rating_col : str
등급 숫자 컬럼 (1=AAA, 2=AA+, ...)
"""
X = df_rated[feature_cols]
y = df_rated[rating_col]
model = sm.OrderedModel(y, X, distr='probit')
result = model.fit(method='bfgs', disp=False)
return result
def predict_shadow_rating(model_result, df_unrated: pd.DataFrame,
feature_cols: list) -> pd.DataFrame:
"""등급 미보유 기업에 Shadow Rating 부여"""
X = df_unrated[feature_cols]
pred_probs = model_result.predict(X)
# 각 기업의 최대 확률 등급
df_unrated = df_unrated.copy()
df_unrated['shadow_rating_numeric'] = pred_probs.values.argmax(axis=1) + 1
return df_unrated
```
### 7.4 등급별 부도율 산출 (블렌딩)
```python
def compute_blended_default_rates(df: pd.DataFrame,
rating_col: str,
default_col: str,
global_dr: dict,
threshold: int = 50) -> pd.DataFrame:
"""
등급별 부도율을 한국 관측 + 글로벌 벤치마크 블렌딩으로 산출.
Parameters
----------
df : pd.DataFrame
전체 기업 데이터 (등급 + 부도여부 포함)
rating_col : str
등급 컬럼명
default_col : str
부도 여부 컬럼명 (0/1)
global_dr : dict
{등급: 글로벌 부도율} 매핑
threshold : int
블렌딩 전환 표본수 임계치
"""
results = []
for grade in sorted(df[rating_col].unique()):
subset = df[df[rating_col] == grade]
n = len(subset)
d = subset[default_col].sum()
kr_dr = d / n if n > 0 else 0
g_dr = global_dr.get(grade, kr_dr)
w = min(1.0, n / threshold)
blended = w * kr_dr + (1 - w) * g_dr
results.append({
'grade': grade,
'n_firms': n,
'n_defaults': d,
'korean_dr': kr_dr,
'global_dr': g_dr,
'weight_kr': w,
'blended_dr': blended
})
return pd.DataFrame(results)
```
### 7.5 베이지안 부도율 추정
```python
from scipy.stats import beta as beta_dist
def bayesian_default_rate(n: int, d: int,
prior_mean: float,
prior_strength: float = 50) -> dict:
"""
베이지안 방식 등급별 부도율 추정.
Parameters
----------
n : int
관측 기업수
d : int
부도 건수
prior_mean : float
사전 부도율 (글로벌 벤치마크)
prior_strength : float
사전 강도 (글로벌 표본수에 비례)
"""
# Beta prior 파라미터
alpha_0 = prior_mean * prior_strength
beta_0 = (1 - prior_mean) * prior_strength
# 사후 분포 (Beta-Binomial conjugacy)
alpha_post = alpha_0 + d
beta_post = beta_0 + (n - d)
# 사후 통계량
posterior_mean = alpha_post / (alpha_post + beta_post)
posterior_mode = (alpha_post - 1) / (alpha_post + beta_post - 2) \
if (alpha_post > 1 and beta_post > 1) else posterior_mean
ci_lower, ci_upper = beta_dist.ppf([0.025, 0.975], alpha_post, beta_post)
return {
'posterior_mean': posterior_mean,
'posterior_mode': posterior_mode,
'ci_95_lower': ci_lower,
'ci_95_upper': ci_upper,
'prior_mean': prior_mean,
'n_obs': n,
'n_defaults': d
}
```
---
## 8. 검증 방법론
### 8.1 변별력(Discriminatory Power) 검증
| 지표 | 설명 | 기준 |
|------|------|------|
| **AUROC** | ROC 곡선 하 면적 | > 0.70 (수용), > 0.80 (양호) |
| **KS 통계량** | 부도/비부도 분포 최대 이격 | > 0.30 (수용) |
| **CAP/AR** | 정확도 비율 | > 0.50 (수용) |
| **정보값(IV)** | 변수별 변별 기여도 | > 0.10 (유의미) |
### 8.2 보정력(Calibration) 검증
| 지표 | 설명 |
|------|------|
| **Hosmer-Lemeshow** | 예측 부도율 vs 실제 부도율 적합도 |
| **Binomial Test** | 등급별 실제 부도율이 예측 구간 내 존재 여부 |
| **Traffic Light** | Basel II 권장 — 녹색/황색/적색 신호 |
### 8.3 백테스팅 프로세스
```
for year in [T-5, T-4, T-3, T-2, T-1]:
1. year 말 기준 DD/EDF 산출
2. year+1 동안의 실제 부도 여부 관측
3. 예측 EDF vs 실현 부도율 비교
4. 변별력/보정력 지표 산출
```
### 8.4 CRA 발표 데이터와 비교
- 한기평·한신평·나이스 연간 발표 등급별 부도율과 본 모형 산출치 비교
- 등급별 편차(bias) 및 상관관계 분석
- 부도 시점 대비 DD 하락 시점의 선행성 분석
---
## 9. 한국 시장 특수 고려사항
### 9.1 데이터 관련
| 항목 | 고려사항 | 대응 |
|------|----------|------|
| KOSPI vs KOSDAQ | KOSDAQ 소형·고변동성 기업 다수 | 시장별 분리 분석 또는 통합+더미 |
| 금융업 | 부채 구조 상이 (예금 = 부채) | **분석 제외** 또는 별도 모형 |
| 등급 불일치 | 한기평·한신평·나이스 등급 차이 | 중위값 또는 최빈값 사용 |
| 분기 vs 연간 | 재무제표 시차 | 분기 데이터 우선, 없으면 연간 보간 |
| 상장폐지 | 부도 vs 비부도 폐지 구분 | 폐지 사유 코드로 필터링 |
### 9.2 구조적 특성
| 특성 | 영향 | 모형 반영 방법 |
|------|------|---------------|
| **재벌 계열** | 그룹 지원으로 개별 DD 대비 부도율 하락 | 계열사 더미 / 그룹 DD 산출 |
| **정부 지원** | 공기업 부도율 ≈ 0 | 정부지원 등급에서 제외 또는 별도 처리 |
| **채권단 자율협약** | 형식적 부도 회피 | 워크아웃 개시를 부도 사건에 포함 |
| **유상증자/CB** | 부도 직전 자본 확충으로 DD 왜곡 | 이벤트 전 DD 사용 또는 플래그 |
### 9.3 변동성 추정 주의사항
| 상황 | 문제 | 대응 |
|------|------|------|
| 장기 거래정지 | 변동성 과소추정 | 정지 기간 제외, 30일 이상 정지 시 분석 제외 |
| 저거래량 | 비유동성 프리미엄 혼재 | 거래량 하위 10% 제외 또는 유동성 보정 |
| 극단 이벤트 | 일시적 급등락으로 변동성 왜곡 | Winsorization (상하 1%) 또는 트리밍 |
| 공매도 제한 | 하락 변동성 억제 | 변동성 하향 편의 인지, 글로벌 대비 보정 |
---
## 10. 기술 스택 및 의존성
### 10.1 Python 패키지
```
# 핵심 (requirements.txt)
numpy>=1.24
pandas>=2.0
scipy>=1.10
statsmodels>=0.14
# 데이터 수집
pykrx>=1.0 # KRX 주가 데이터
OpenDartReader>=0.3 # DART 전자공시 API
# 변동성 모형
arch>=6.0 # GARCH/EWMA 모형
# 시각화
matplotlib>=3.7
plotly>=5.15
seaborn>=0.12
# 머신러닝 (선택)
scikit-learn>=1.3
xgboost>=2.0
# 유틸리티
tqdm>=4.65
pyyaml>=6.0
```
### 10.2 무위험이자율
- 한국은행 ECOS의 **국고채 1년물 금리** 사용
- 또는 통화안정증권(MSB) 1년물
### 10.3 데이터베이스
- 개발 단계: SQLite (로컬 파일)
- 운영 단계: PostgreSQL (필요시)
---
## 11. 참고 문헌
### 핵심 논문
1. Merton, R.C. (1974). "On the Pricing of Corporate Debt: The Risk Structure of Interest Rates." *Journal of Finance*, 29(2), 449-470.
2. Black, F. & Cox, J.C. (1976). "Valuing Corporate Securities: Some Effects of Bond Indenture Provisions." *Journal of Finance*, 31(2), 351-367.
3. Bharath, S.T. & Shumway, T. (2008). "Forecasting Default with the Merton Distance to Default Model." *Review of Financial Studies*, 21(3), 1339-1369.
4. Crosbie, P. & Bohn, J. (2003). "Modeling Default Risk." Moody's KMV Working Paper.
### 한국 시장 연구
5. 한국은행. "IRB 접근법 하에서의 장기 부도확률 추정."
6. 한국기업평가. "연간 부도율 통계" (korearatings.com)
7. 한국신용평가. "신용등급별 부도율 및 전이행렬" (kisrating.com)
### 기술 참고
8. Credit Suisse Financial Products. (1997). "CreditRisk+: A Credit Risk Management Framework."
9. JP Morgan. (1997). "CreditMetrics — Technical Document."
10. Basel Committee on Banking Supervision. "Studies on the Validation of Internal Rating Systems."

13
requirements.txt Normal file
View File

@@ -0,0 +1,13 @@
numpy>=1.24
pandas>=2.0
scipy>=1.10
statsmodels>=0.14
pykrx>=1.0
opendart-reader>=0.3
arch>=6.0
matplotlib>=3.7
plotly>=5.15
seaborn>=0.12
scikit-learn>=1.3
tqdm>=4.65
pyyaml>=6.0

1
src/__init__.py Normal file
View File

@@ -0,0 +1 @@
# src package

1
src/data/__init__.py Normal file
View File

@@ -0,0 +1 @@
# data package

315
src/data/dart_fetcher.py Normal file
View File

@@ -0,0 +1,315 @@
"""
DART 재무제표 데이터 수집 모듈
OpenDartReader를 사용하여 KRX 상장사의 재무제표(부채구조)를 수집합니다.
Merton 모형에 필요한 유동부채(STD), 비유동부채(LTD), 총자산 등을 추출합니다.
"""
import os
import time
import yaml
import pandas as pd
import numpy as np
from datetime import datetime
from pathlib import Path
from tqdm import tqdm
try:
import OpenDartReader
except ImportError:
raise ImportError("OpenDartReader가 설치되지 않았습니다. pip install opendart-reader 를 실행하세요.")
def load_config() -> dict:
"""config/settings.yaml 로드"""
config_path = Path(__file__).parent.parent.parent / "config" / "settings.yaml"
with open(config_path, "r", encoding="utf-8") as f:
return yaml.safe_load(f)
def init_dart(api_key: str = None) -> OpenDartReader:
"""OpenDartReader 초기화"""
if api_key is None:
config = load_config()
api_key = config["dart_api_key"]
return OpenDartReader(api_key)
def get_corp_codes(dart: OpenDartReader) -> pd.DataFrame:
"""
DART 기업 코드 목록 조회 (상장사)
Returns
-------
pd.DataFrame
columns: [corp_code, corp_name, stock_code, modify_date]
"""
corp_list = dart.corp_codes
# 상장사만 (stock_code가 존재하는 것)
listed = corp_list[corp_list["stock_code"].notna() & (corp_list["stock_code"] != " ")]
return listed.reset_index(drop=True)
def extract_financial_items(dart: OpenDartReader,
corp_code: str,
year: int,
report_code: str = "11011") -> dict:
"""
특정 기업의 재무제표에서 Merton 모형에 필요한 항목 추출.
Parameters
----------
dart : OpenDartReader
corp_code : str
DART 고유 기업 코드
year : int
사업연도
report_code : str
보고서 코드 (11011=사업보고서/연간, 11012=반기, 11013=1분기, 11014=3분기)
Returns
-------
dict with keys:
total_assets, current_liabilities (STD), non_current_liabilities (LTD),
total_liabilities, total_equity, revenue, operating_income,
interest_expense, net_income, ebitda_proxy
"""
result = {
"total_assets": np.nan,
"current_liabilities": np.nan, # 유동부채 = STD
"non_current_liabilities": np.nan, # 비유동부채 = LTD
"total_liabilities": np.nan,
"total_equity": np.nan,
"revenue": np.nan,
"operating_income": np.nan,
"interest_expense": np.nan,
"net_income": np.nan,
}
try:
# 재무상태표 (BS)
bs = dart.finstate(corp_code, year, reprt_code=report_code)
if bs is None or len(bs) == 0:
return result
# 연결재무제표 우선, 없으면 개별
if "fs_div" in bs.columns:
cfs = bs[bs["fs_div"] == "CFS"] # 연결
if len(cfs) == 0:
cfs = bs[bs["fs_div"] == "OFS"] # 개별
else:
cfs = bs
if len(cfs) == 0:
return result
# 항목명으로 검색하는 헬퍼
def find_amount(df, keywords, col="thstrm_amount"):
"""키워드 목록으로 금액 검색"""
if col not in df.columns:
# 대체 컬럼 시도
for alt_col in ["thstrm_amount", "frmtrm_amount", "bfefrmtrm_amount"]:
if alt_col in df.columns:
col = alt_col
break
else:
return np.nan
for kw in keywords:
matches = df[df["account_nm"].str.contains(kw, na=False, regex=False)]
if len(matches) > 0:
val = matches.iloc[0][col]
if pd.notna(val):
# 쉼표 제거 후 숫자 변환
if isinstance(val, str):
val = val.replace(",", "").replace(" ", "")
try:
return float(val)
except ValueError:
return np.nan
return float(val)
return np.nan
# 재무상태표 항목 추출
result["total_assets"] = find_amount(cfs, ["자산총계", "자산 총계"])
result["current_liabilities"] = find_amount(cfs, ["유동부채", "유동 부채"])
result["non_current_liabilities"] = find_amount(cfs, ["비유동부채", "비유동 부채"])
result["total_liabilities"] = find_amount(cfs, ["부채총계", "부채 총계"])
result["total_equity"] = find_amount(cfs, ["자본총계", "자본 총계"])
# 손익계산서 항목
result["revenue"] = find_amount(cfs, ["매출액", "매출", "영업수익", "수익(매출액)"])
result["operating_income"] = find_amount(cfs, ["영업이익", "영업 이익"])
result["net_income"] = find_amount(cfs, ["당기순이익", "당기 순이익", "분기순이익"])
result["interest_expense"] = find_amount(cfs, ["이자비용", "금융비용", "금융원가"])
except Exception as e:
result["_error"] = str(e)
return result
def compute_derived_ratios(row: dict) -> dict:
"""
Merton 모형 및 Shadow Rating에 필요한 파생 비율 계산
"""
derived = {}
# 부도점 (Default Point) = STD + 0.5 * LTD
std = row.get("current_liabilities", np.nan)
ltd = row.get("non_current_liabilities", np.nan)
if pd.notna(std) and pd.notna(ltd):
derived["default_point"] = std + 0.5 * ltd
else:
derived["default_point"] = np.nan
# 레버리지 비율 = 총부채 / 총자산
ta = row.get("total_assets", np.nan)
tl = row.get("total_liabilities", np.nan)
if pd.notna(tl) and pd.notna(ta) and ta > 0:
derived["leverage_ratio"] = tl / ta
else:
derived["leverage_ratio"] = np.nan
# 이자보상비율 = 영업이익 / 이자비용
oi = row.get("operating_income", np.nan)
ie = row.get("interest_expense", np.nan)
if pd.notna(oi) and pd.notna(ie) and ie > 0:
derived["interest_coverage"] = oi / ie
else:
derived["interest_coverage"] = np.nan
# ROA = 순이익 / 총자산
ni = row.get("net_income", np.nan)
if pd.notna(ni) and pd.notna(ta) and ta > 0:
derived["roa"] = ni / ta
else:
derived["roa"] = np.nan
# 유동비율 = (총자산 - 비유동자산 근사) / 유동부채
# 간접 산출: 유동자산 ≈ 총자산 - (총부채 - 유동부채 + 자본 - 유동자산)
# 단순화: current_ratio = (총자산 - 비유동부채 근사) 는 어려우므로,
# 유동부채 대비 총자산 비율로 대체
if pd.notna(ta) and pd.notna(std) and std > 0:
derived["asset_to_std_ratio"] = ta / std
else:
derived["asset_to_std_ratio"] = np.nan
# 기업 규모 (로그 총자산)
if pd.notna(ta) and ta > 0:
derived["log_assets"] = np.log(ta)
else:
derived["log_assets"] = np.nan
return derived
def fetch_all_financial_data(tickers: list,
year: int = 2024,
config: dict = None) -> pd.DataFrame:
"""
전 상장사의 재무제표 데이터를 수집.
Parameters
----------
tickers : list
종목코드(6자리) 리스트
year : int
사업연도
config : dict
설정 딕셔너리
Returns
-------
pd.DataFrame
종목별 재무 데이터 + 파생 비율
"""
if config is None:
config = load_config()
dart = init_dart(config["dart_api_key"])
sleep_sec = config.get("krx", {}).get("sleep_seconds", 0.5)
# DART 기업코드 목록
print("[DART] 기업코드 목록 로딩...")
corp_codes = get_corp_codes(dart)
# ticker → corp_code 매핑
ticker_to_corp = {}
for _, row in corp_codes.iterrows():
sc = str(row["stock_code"]).strip()
if sc and sc != "nan":
ticker_to_corp[sc] = row["corp_code"]
print(f"[DART] {len(ticker_to_corp)}개 상장사 매핑 완료")
records = []
errors = []
print(f"[DART] {year}년 재무제표 수집 중...")
for ticker in tqdm(tickers, desc="재무제표 수집"):
ticker_str = str(ticker).zfill(6)
corp_code = ticker_to_corp.get(ticker_str)
if corp_code is None:
errors.append({"ticker": ticker_str, "error": "corp_code not found"})
continue
try:
fin_data = extract_financial_items(dart, corp_code, year)
derived = compute_derived_ratios(fin_data)
record = {"ticker": ticker_str, "year": year}
record.update(fin_data)
record.update(derived)
records.append(record)
time.sleep(sleep_sec) # Rate limiting
except Exception as e:
errors.append({"ticker": ticker_str, "error": str(e)})
time.sleep(sleep_sec)
continue
df = pd.DataFrame(records)
if "ticker" in df.columns:
df = df.set_index("ticker")
print(f"[DART] 수집 완료: {len(df)}개 (에러: {len(errors)}건)")
return df, errors
def save_financial_data(df: pd.DataFrame, errors: list,
year: int, output_dir: str = None):
"""수집 결과를 CSV로 저장"""
if output_dir is None:
output_dir = Path(__file__).parent.parent.parent / "data" / "raw"
output_dir = Path(output_dir)
output_dir.mkdir(parents=True, exist_ok=True)
df.to_csv(output_dir / f"financial_data_{year}.csv", encoding="utf-8-sig")
if errors:
pd.DataFrame(errors).to_csv(
output_dir / f"financial_errors_{year}.csv", encoding="utf-8-sig", index=False
)
print(f"[DART] 데이터 저장 완료: {output_dir}")
# ---- CLI 실행 ----
if __name__ == "__main__":
from krx_fetcher import load_config, get_market_cap_all
config = load_config()
year = config.get("end_year", 2024) - 1 # 가장 최근 확정 사업연도
# 상장사 목록 가져오기
target_date = datetime.now().strftime("%Y%m%d")
market_cap = get_market_cap_all(target_date)
tickers = list(market_cap.index)
# 재무 데이터 수집
df, errors = fetch_all_financial_data(tickers, year=year, config=config)
save_financial_data(df, errors, year)

311
src/data/database.py Normal file
View File

@@ -0,0 +1,311 @@
"""
EDF 프로젝트 SQLite 데이터베이스 모듈
KRX 주가, DART 재무제표, Merton DD/EDF 결과를 영구 저장합니다.
"""
import sqlite3
import pandas as pd
import numpy as np
from pathlib import Path
from datetime import datetime
DB_PATH = Path(__file__).parent.parent.parent / "data" / "edf.db"
def get_connection(db_path: str = None) -> sqlite3.Connection:
"""SQLite 연결 반환"""
if db_path is None:
db_path = str(DB_PATH)
Path(db_path).parent.mkdir(parents=True, exist_ok=True)
conn = sqlite3.connect(db_path)
conn.execute("PRAGMA journal_mode=WAL")
conn.execute("PRAGMA foreign_keys=ON")
return conn
def init_db(conn: sqlite3.Connection = None):
"""데이터베이스 스키마 초기화"""
if conn is None:
conn = get_connection()
conn.executescript("""
-- ============================================================
-- 1. 종목 마스터
-- ============================================================
CREATE TABLE IF NOT EXISTS companies (
ticker TEXT PRIMARY KEY, -- 종목코드 (6자리)
name TEXT, -- 종목명
market TEXT, -- KOSPI / KOSDAQ
sector TEXT, -- 업종
corp_code TEXT, -- DART 고유코드
is_financial INTEGER DEFAULT 0, -- 금융업 여부 (1=금융, 0=비금융)
updated_at TEXT
);
-- ============================================================
-- 2. 일별 시장 데이터 (주가, 시가총액)
-- ============================================================
CREATE TABLE IF NOT EXISTS market_data (
ticker TEXT NOT NULL,
date TEXT NOT NULL, -- YYYY-MM-DD
close_price REAL, -- 종가 (원)
market_cap REAL, -- 시가총액 (원)
volume INTEGER, -- 거래량
shares INTEGER, -- 상장주식수
PRIMARY KEY (ticker, date),
FOREIGN KEY (ticker) REFERENCES companies(ticker)
);
CREATE INDEX IF NOT EXISTS idx_market_data_date ON market_data(date);
-- ============================================================
-- 3. 재무제표 데이터 (연간/분기)
-- ============================================================
CREATE TABLE IF NOT EXISTS financial_data (
ticker TEXT NOT NULL,
year INTEGER NOT NULL, -- 사업연도
report_type TEXT DEFAULT 'annual', -- annual / q1 / q2 / q3
total_assets REAL, -- 총자산
current_liabilities REAL, -- 유동부채 (STD)
non_current_liabilities REAL, -- 비유동부채 (LTD)
total_liabilities REAL, -- 부채총계
total_equity REAL, -- 자본총계
revenue REAL, -- 매출액
operating_income REAL, -- 영업이익
net_income REAL, -- 당기순이익
interest_expense REAL, -- 이자비용
-- 파생 항목 (계산됨)
default_point REAL, -- STD + 0.5*LTD
leverage_ratio REAL, -- 총부채/총자산
interest_coverage REAL, -- 영업이익/이자비용
roa REAL, -- 순이익/총자산
log_assets REAL, -- ln(총자산)
fetched_at TEXT, -- 수집 시각
PRIMARY KEY (ticker, year, report_type),
FOREIGN KEY (ticker) REFERENCES companies(ticker)
);
-- ============================================================
-- 4. 변동성 산출 결과
-- ============================================================
CREATE TABLE IF NOT EXISTS volatility (
ticker TEXT NOT NULL,
base_date TEXT NOT NULL, -- 기준일 (YYYY-MM-DD)
method TEXT NOT NULL, -- historical / ewma / garch
sigma_E REAL, -- 주가변동성 (연환산)
n_trading_days INTEGER, -- 사용 거래일수
PRIMARY KEY (ticker, base_date, method),
FOREIGN KEY (ticker) REFERENCES companies(ticker)
);
-- ============================================================
-- 5. Merton DD/EDF 결과
-- ============================================================
CREATE TABLE IF NOT EXISTS merton_results (
ticker TEXT NOT NULL,
base_date TEXT NOT NULL, -- 기준일
fin_year INTEGER NOT NULL, -- 사용된 재무 연도
E REAL, -- 자기자본 시장가치
sigma_E REAL, -- 주가변동성
D REAL, -- 부도점
V REAL, -- 추정 자산가치
sigma_V REAL, -- 추정 자산변동성
DD REAL, -- Distance-to-Default
EDF REAL, -- Expected Default Frequency
leverage REAL, -- D/V
method TEXT, -- fsolve / iterative / naive_fallback
dd_rating TEXT, -- DD 기반 내재등급
PRIMARY KEY (ticker, base_date),
FOREIGN KEY (ticker) REFERENCES companies(ticker)
);
CREATE INDEX IF NOT EXISTS idx_merton_dd ON merton_results(DD);
CREATE INDEX IF NOT EXISTS idx_merton_rating ON merton_results(dd_rating);
-- ============================================================
-- 6. 신용등급 (실제 관측 등급)
-- ============================================================
CREATE TABLE IF NOT EXISTS credit_ratings (
ticker TEXT NOT NULL,
rating_date TEXT NOT NULL, -- 등급 확인일
agency TEXT, -- 한기평/한신평/나이스
rating TEXT, -- AAA, AA+, ... , D
source TEXT, -- DART공시 / 수동입력
PRIMARY KEY (ticker, rating_date, agency),
FOREIGN KEY (ticker) REFERENCES companies(ticker)
);
-- ============================================================
-- 7. 부도 이력
-- ============================================================
CREATE TABLE IF NOT EXISTS defaults (
ticker TEXT NOT NULL,
default_date TEXT NOT NULL, -- 부도 발생일
default_type TEXT, -- 법정관리/워크아웃/상장폐지/부도어음
description TEXT,
PRIMARY KEY (ticker, default_date),
FOREIGN KEY (ticker) REFERENCES companies(ticker)
);
-- ============================================================
-- 8. 등급별 부도율 최종 결과
-- ============================================================
CREATE TABLE IF NOT EXISTS default_rates (
base_date TEXT NOT NULL, -- 산출 기준일
rating_grade TEXT NOT NULL, -- 등급
n_firms INTEGER, -- 관측 기업수
n_defaults INTEGER, -- 부도 건수
korean_dr REAL, -- 한국 관측 부도율
global_dr REAL, -- 글로벌 벤치마크 부도율
weight_kr REAL, -- 한국 가중치
blended_dr REAL, -- 블렌딩 부도율
bayesian_dr REAL, -- 베이지안 사후 부도율
PRIMARY KEY (base_date, rating_grade)
);
-- ============================================================
-- 메타 정보
-- ============================================================
CREATE TABLE IF NOT EXISTS meta (
key TEXT PRIMARY KEY,
value TEXT
);
""")
# 메타 정보 기록
conn.execute(
"INSERT OR REPLACE INTO meta (key, value) VALUES (?, ?)",
("schema_version", "1.0")
)
conn.execute(
"INSERT OR REPLACE INTO meta (key, value) VALUES (?, ?)",
("created_at", datetime.now().isoformat())
)
conn.commit()
print(f"[DB] Schema initialized: {conn.execute('SELECT value FROM meta WHERE key=?', ('schema_version',)).fetchone()[0]}")
return conn
# ============================================================
# DataFrame ↔ SQLite 유틸리티
# ============================================================
def upsert_companies(conn: sqlite3.Connection, df: pd.DataFrame):
"""종목 마스터 upsert"""
now = datetime.now().isoformat()
for _, row in df.iterrows():
conn.execute("""
INSERT OR REPLACE INTO companies (ticker, name, market, sector, corp_code, updated_at)
VALUES (?, ?, ?, ?, ?, ?)
""", (
str(row.get("ticker", "")),
str(row.get("name", "")),
str(row.get("market", "")),
str(row.get("sector", "")),
str(row.get("corp_code", "")),
now
))
conn.commit()
def upsert_market_data(conn: sqlite3.Connection, ticker: str, df: pd.DataFrame):
"""일별 시장 데이터 upsert (df.index = DatetimeIndex)"""
records = []
for date, row in df.iterrows():
date_str = date.strftime("%Y-%m-%d") if hasattr(date, 'strftime') else str(date)
records.append((
ticker, date_str,
float(row.get("종가", 0)),
float(row.get("시가총액", 0)) if "시가총액" in row else None,
int(row.get("거래량", 0)),
int(row.get("상장주식수", 0)) if "상장주식수" in row else None,
))
conn.executemany("""
INSERT OR REPLACE INTO market_data (ticker, date, close_price, market_cap, volume, shares)
VALUES (?, ?, ?, ?, ?, ?)
""", records)
conn.commit()
def upsert_financial(conn: sqlite3.Connection, df: pd.DataFrame):
"""재무제표 데이터 upsert"""
now = datetime.now().isoformat()
for idx, row in df.iterrows():
ticker = str(idx) if isinstance(idx, str) else str(row.get("ticker", idx))
conn.execute("""
INSERT OR REPLACE INTO financial_data
(ticker, year, report_type, total_assets, current_liabilities, non_current_liabilities,
total_liabilities, total_equity, revenue, operating_income, net_income, interest_expense,
default_point, leverage_ratio, interest_coverage, roa, log_assets, fetched_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
ticker,
int(row.get("year", 0)),
str(row.get("report_type", "annual")),
row.get("total_assets"),
row.get("current_liabilities"),
row.get("non_current_liabilities"),
row.get("total_liabilities"),
row.get("total_equity"),
row.get("revenue"),
row.get("operating_income"),
row.get("net_income"),
row.get("interest_expense"),
row.get("default_point"),
row.get("leverage_ratio"),
row.get("interest_coverage"),
row.get("roa"),
row.get("log_assets"),
now
))
conn.commit()
def upsert_merton_results(conn: sqlite3.Connection, df: pd.DataFrame, base_date: str, fin_year: int):
"""Merton DD/EDF 결과 upsert"""
for idx, row in df.iterrows():
ticker = str(idx)
conn.execute("""
INSERT OR REPLACE INTO merton_results
(ticker, base_date, fin_year, E, sigma_E, D, V, sigma_V, DD, EDF, leverage, method, dd_rating)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
ticker, base_date, fin_year,
row.get("E"), row.get("sigma_E"), row.get("D"),
row.get("V"), row.get("sigma_V"),
row.get("DD"), row.get("EDF"), row.get("leverage"),
row.get("method"), row.get("dd_rating")
))
conn.commit()
def load_merton_results(conn: sqlite3.Connection, base_date: str = None) -> pd.DataFrame:
"""Merton 결과 조회"""
if base_date:
query = "SELECT * FROM merton_results WHERE base_date = ?"
return pd.read_sql_query(query, conn, params=(base_date,), index_col="ticker")
else:
query = "SELECT * FROM merton_results"
return pd.read_sql_query(query, conn, index_col="ticker")
def get_stats(conn: sqlite3.Connection) -> dict:
"""DB 통계 조회"""
stats = {}
for table in ["companies", "market_data", "financial_data", "volatility",
"merton_results", "credit_ratings", "defaults", "default_rates"]:
try:
count = conn.execute(f"SELECT COUNT(*) FROM {table}").fetchone()[0]
stats[table] = count
except Exception:
stats[table] = 0
return stats
# ---- CLI ----
if __name__ == "__main__":
conn = init_db()
stats = get_stats(conn)
print("\n[DB] Table stats:")
for table, count in stats.items():
print(f" {table}: {count} rows")
conn.close()

261
src/data/krx_fetcher.py Normal file
View File

@@ -0,0 +1,261 @@
"""
KRX 주가/시가총액 데이터 수집 모듈
pykrx를 사용하여 KRX 상장사의 일별 주가, 시가총액, 거래량 데이터를 수집합니다.
"""
import os
import time
import yaml
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from pathlib import Path
from tqdm import tqdm
try:
from pykrx import stock
except ImportError:
raise ImportError("pykrx가 설치되지 않았습니다. pip install pykrx 를 실행하세요.")
def load_config() -> dict:
"""config/settings.yaml 로드"""
config_path = Path(__file__).parent.parent.parent / "config" / "settings.yaml"
with open(config_path, "r", encoding="utf-8") as f:
return yaml.safe_load(f)
def get_all_tickers(date: str, market: str = "ALL") -> pd.DataFrame:
"""
특정 날짜의 전 종목 티커/종목명 조회
Parameters
----------
date : str
조회 날짜 (YYYYMMDD)
market : str
시장 구분 (KOSPI, KOSDAQ, ALL)
Returns
-------
pd.DataFrame
columns: [ticker, name, market]
"""
tickers = stock.get_market_ticker_list(date, market=market)
records = []
for t in tickers:
name = stock.get_market_ticker_name(t)
records.append({"ticker": t, "name": name, "market": market})
return pd.DataFrame(records)
def get_market_cap_all(date: str, market: str = "ALL") -> pd.DataFrame:
"""
특정 날짜의 전 종목 시가총액 조회
Returns
-------
pd.DataFrame
index: ticker, columns: [종가, 시가총액, 거래량, 거래대금, 상장주식수]
"""
df = stock.get_market_cap(date, market=market)
df.index.name = "ticker"
return df
def get_ohlcv(ticker: str, start: str, end: str) -> pd.DataFrame:
"""
특정 종목의 일별 OHLCV 데이터 조회
Parameters
----------
ticker : str
종목코드 (6자리)
start : str
시작 날짜 (YYYYMMDD)
end : str
종료 날짜 (YYYYMMDD)
Returns
-------
pd.DataFrame
columns: [시가, 고가, 저가, 종가, 거래량]
"""
df = stock.get_market_ohlcv(start, end, ticker)
return df
def calculate_equity_volatility(prices: pd.Series,
method: str = "historical",
window: int = 252,
ewma_lambda: float = 0.94) -> float:
"""
주가 수익률 변동성 추정 (연환산)
Parameters
----------
prices : pd.Series
일별 종가 시리즈
method : str
추정 방법 (historical / ewma / garch)
window : int
추정 윈도우 (거래일)
ewma_lambda : float
EWMA lambda (method='ewma'일 때)
Returns
-------
float
연환산 변동성
"""
# 로그 수익률
log_returns = np.log(prices / prices.shift(1)).dropna()
if len(log_returns) < 30:
return np.nan
if method == "historical":
use_returns = log_returns.tail(window) if len(log_returns) >= window else log_returns
return float(use_returns.std() * np.sqrt(252))
elif method == "ewma":
variance = log_returns.iloc[0] ** 2
for ret in log_returns.iloc[1:]:
variance = ewma_lambda * variance + (1 - ewma_lambda) * ret ** 2
return float(np.sqrt(variance * 252))
elif method == "garch":
try:
from arch import arch_model
returns_pct = log_returns * 100
model = arch_model(returns_pct, vol='Garch', p=1, q=1, dist='normal')
result = model.fit(disp='off', show_warning=False)
cond_vol = result.conditional_volatility.iloc[-1] / 100
return float(cond_vol * np.sqrt(252))
except Exception:
# GARCH 실패 시 historical로 폴백
return calculate_equity_volatility(prices, "historical", window)
else:
raise ValueError(f"Unknown method: {method}")
def fetch_all_stock_data(target_date: str = None,
lookback_years: int = 2,
config: dict = None) -> dict:
"""
전 종목의 주가 데이터 및 변동성을 수집.
Parameters
----------
target_date : str
기준 날짜 (YYYYMMDD). None이면 최근 거래일.
lookback_years : int
주가 수집 기간 (년)
config : dict
설정 딕셔너리
Returns
-------
dict with keys:
'market_cap': 기준일 시가총액 DataFrame
'volatility': 종목별 변동성 DataFrame
'tickers': 종목 정보 DataFrame
"""
if config is None:
config = load_config()
if target_date is None:
target_date = datetime.now().strftime("%Y%m%d")
sleep_sec = config.get("krx", {}).get("sleep_seconds", 0.5)
vol_method = config.get("merton", {}).get("volatility_method", "historical")
vol_window = config.get("merton", {}).get("volatility_window", 252)
ewma_lam = config.get("merton", {}).get("ewma_lambda", 0.94)
min_trading = config.get("krx", {}).get("min_trading_days", 200)
# 시작일 계산
target_dt = datetime.strptime(target_date, "%Y%m%d")
start_dt = target_dt - timedelta(days=365 * lookback_years + 30)
start_date = start_dt.strftime("%Y%m%d")
print(f"[KRX] 기준일: {target_date}, 주가 수집 시작일: {start_date}")
# 1) 시가총액 조회
print("[KRX] 시가총액 조회 중...")
market_cap = get_market_cap_all(target_date)
time.sleep(sleep_sec)
# 2) 종목 리스트
print(f"[KRX] 총 {len(market_cap)}개 종목 확인")
# 3) 변동성 산출
vol_records = []
errors = []
print(f"[KRX] 종목별 변동성 산출 중 (method={vol_method})...")
for ticker in tqdm(market_cap.index, desc="변동성 산출"):
try:
ohlcv = get_ohlcv(ticker, start_date, target_date)
time.sleep(sleep_sec)
if len(ohlcv) < min_trading:
continue
prices = ohlcv["종가"]
prices = prices[prices > 0] # 0원 제거
if len(prices) < min_trading:
continue
sigma_e = calculate_equity_volatility(
prices, method=vol_method, window=vol_window, ewma_lambda=ewma_lam
)
vol_records.append({
"ticker": ticker,
"sigma_E": sigma_e,
"n_trading_days": len(prices),
"last_price": float(prices.iloc[-1]),
})
except Exception as e:
errors.append({"ticker": ticker, "error": str(e)})
continue
vol_df = pd.DataFrame(vol_records).set_index("ticker")
print(f"[KRX] 변동성 산출 완료: {len(vol_df)}개 종목 (에러: {len(errors)}건)")
return {
"market_cap": market_cap,
"volatility": vol_df,
"errors": errors,
}
def save_stock_data(data: dict, output_dir: str = None):
"""수집 결과를 CSV로 저장"""
if output_dir is None:
output_dir = Path(__file__).parent.parent.parent / "data" / "raw"
output_dir = Path(output_dir)
output_dir.mkdir(parents=True, exist_ok=True)
timestamp = datetime.now().strftime("%Y%m%d")
data["market_cap"].to_csv(output_dir / f"market_cap_{timestamp}.csv", encoding="utf-8-sig")
data["volatility"].to_csv(output_dir / f"volatility_{timestamp}.csv", encoding="utf-8-sig")
if data.get("errors"):
pd.DataFrame(data["errors"]).to_csv(
output_dir / f"fetch_errors_{timestamp}.csv", encoding="utf-8-sig", index=False
)
print(f"[KRX] 데이터 저장 완료: {output_dir}")
# ---- CLI 실행 ----
if __name__ == "__main__":
config = load_config()
data = fetch_all_stock_data(config=config)
save_stock_data(data)

1
src/models/__init__.py Normal file
View File

@@ -0,0 +1 @@
# models package

371
src/models/merton.py Normal file
View File

@@ -0,0 +1,371 @@
"""
Merton-KMV 모형 모듈
기업의 자기자본 시장가치와 주가변동성으로부터
자산가치/자산변동성을 추정하고 Distance-to-Default(DD) 및 EDF를 산출합니다.
"""
import numpy as np
import pandas as pd
from scipy.optimize import fsolve, brentq
from scipy.stats import norm
from pathlib import Path
import yaml
def solve_merton(E: float, sigma_E: float, D: float,
r: float, T: float = 1.0,
max_iter: int = 100, tol: float = 1e-6) -> dict:
"""
Merton 연립방정식을 풀어 자산가치(V)와 자산변동성(σ_V)을 반복 추정.
두 가지 방법을 시도:
1) scipy.fsolve (뉴턴법)
2) 실패 시 반복 대입법 (Iterative substitution)
Parameters
----------
E : float
자기자본 시장가치 (시가총액, 원 단위)
sigma_E : float
주가수익률 변동성 (연환산, 예: 0.30)
D : float
부도점 (= STD + 0.5 * LTD, 원 단위)
r : float
무위험이자율 (연, 예: 0.035)
T : float
시간 수평선 (년, 기본 1.0)
max_iter : int
반복 대입법 최대 반복 횟수
tol : float
수렴 허용 오차
Returns
-------
dict:
V: 추정 자산가치
sigma_V: 추정 자산변동성
d1, d2: Black-Scholes 파라미터
converged: 수렴 여부
method: 사용된 방법
"""
if E <= 0 or sigma_E <= 0 or D <= 0:
return {"V": np.nan, "sigma_V": np.nan, "d1": np.nan, "d2": np.nan,
"converged": False, "method": "invalid_input"}
# --- 방법 1: fsolve ---
def equations(params):
V, sigma_V = params
if V <= 0 or sigma_V <= 0:
return [1e10, 1e10]
d1 = (np.log(V / D) + (r + 0.5 * sigma_V**2) * T) / (sigma_V * np.sqrt(T))
d2 = d1 - sigma_V * np.sqrt(T)
eq1 = V * norm.cdf(d1) - D * np.exp(-r * T) * norm.cdf(d2) - E
eq2 = (V / E) * norm.cdf(d1) * sigma_V - sigma_E
return [eq1, eq2]
# 초기값
V0 = E + D
sigma_V0 = sigma_E * E / (E + D)
try:
sol, info, ier, msg = fsolve(equations, [V0, sigma_V0], full_output=True)
V_sol, sigma_V_sol = sol
if ier == 1 and V_sol > 0 and sigma_V_sol > 0:
d1 = (np.log(V_sol / D) + (r + 0.5 * sigma_V_sol**2) * T) / (sigma_V_sol * np.sqrt(T))
d2 = d1 - sigma_V_sol * np.sqrt(T)
return {
"V": V_sol, "sigma_V": sigma_V_sol,
"d1": d1, "d2": d2,
"converged": True, "method": "fsolve"
}
except Exception:
pass
# --- 방법 2: 반복 대입법 ---
sigma_V = sigma_V0
V = V0
for i in range(max_iter):
d1 = (np.log(V / D) + (r + 0.5 * sigma_V**2) * T) / (sigma_V * np.sqrt(T))
d2 = d1 - sigma_V * np.sqrt(T)
# V 업데이트: E = V*N(d1) - D*e^(-rT)*N(d2) → V = (E + D*e^(-rT)*N(d2)) / N(d1)
Nd1 = norm.cdf(d1)
Nd2 = norm.cdf(d2)
if Nd1 < 1e-10:
break
V_new = (E + D * np.exp(-r * T) * Nd2) / Nd1
# sigma_V 업데이트: sigma_E = (V/E)*N(d1)*sigma_V → sigma_V = sigma_E*E / (V*N(d1))
sigma_V_new = sigma_E * E / (V_new * Nd1)
if abs(V_new - V) / V < tol and abs(sigma_V_new - sigma_V) / sigma_V < tol:
d1 = (np.log(V_new / D) + (r + 0.5 * sigma_V_new**2) * T) / (sigma_V_new * np.sqrt(T))
d2 = d1 - sigma_V_new * np.sqrt(T)
return {
"V": V_new, "sigma_V": sigma_V_new,
"d1": d1, "d2": d2,
"converged": True, "method": "iterative"
}
V = V_new
sigma_V = sigma_V_new
# 수렴하지 않았지만 마지막 값 반환
d1 = (np.log(V / D) + (r + 0.5 * sigma_V**2) * T) / (sigma_V * np.sqrt(T))
d2 = d1 - sigma_V * np.sqrt(T)
return {
"V": V, "sigma_V": sigma_V,
"d1": d1, "d2": d2,
"converged": False, "method": "iterative_no_converge"
}
def calculate_dd(V: float, sigma_V: float, D: float,
mu: float = None, r: float = 0.035,
T: float = 1.0) -> float:
"""
Distance-to-Default 산출
Parameters
----------
V : float
자산가치
sigma_V : float
자산변동성
D : float
부도점
mu : float
자산 기대수익률 (None이면 r 사용)
r : float
무위험이자율
T : float
시간 수평선
"""
if mu is None:
mu = r
if D <= 0 or V <= 0 or sigma_V <= 0:
return np.nan
DD = (np.log(V / D) + (mu - 0.5 * sigma_V**2) * T) / (sigma_V * np.sqrt(T))
return DD
def calculate_edf(DD: float) -> float:
"""이론적 EDF 산출 (정규분포 가정)"""
if np.isnan(DD):
return np.nan
return norm.cdf(-DD)
def naive_dd(E: float, sigma_E: float, D: float,
mu: float = None, r: float = 0.035,
T: float = 1.0) -> dict:
"""
Bharath-Shumway 간편 DD (반복 추정 없이 직접 산출)
빠른 1차 필터링이나 반복추정 실패 시 대안으로 사용.
"""
if E <= 0 or sigma_E <= 0 or D <= 0:
return {"DD": np.nan, "EDF": np.nan, "V": np.nan, "sigma_V": np.nan}
if mu is None:
mu = r
V = E + D
sigma_V = (E / (E + D)) * sigma_E + (D / (E + D)) * (0.05 + 0.25 * sigma_E)
DD = (np.log(V / D) + (mu - 0.5 * sigma_V**2) * T) / (sigma_V * np.sqrt(T))
EDF = norm.cdf(-DD)
return {"DD": DD, "EDF": EDF, "V": V, "sigma_V": sigma_V}
def run_merton_for_all(market_data: pd.DataFrame,
financial_data: pd.DataFrame,
r: float = 0.035,
T: float = 1.0,
use_naive_fallback: bool = True) -> pd.DataFrame:
"""
전 종목에 대해 Merton 모형 실행.
Parameters
----------
market_data : pd.DataFrame
index=ticker, columns 포함: [시가총액, sigma_E]
(krx_fetcher에서 시가총액 + volatility 병합한 데이터)
financial_data : pd.DataFrame
index=ticker, columns 포함: [default_point, ...]
(dart_fetcher에서 수집한 재무 데이터)
r : float
무위험이자율
T : float
시간 수평선
use_naive_fallback : bool
Merton 수렴 실패 시 Naive DD 사용 여부
Returns
-------
pd.DataFrame
종목별 V, sigma_V, DD, EDF 등
"""
# 두 데이터셋 병합
common_tickers = market_data.index.intersection(financial_data.index)
print(f"[Merton] 공통 종목 수: {len(common_tickers)}")
results = []
for ticker in common_tickers:
mkt = market_data.loc[ticker]
fin = financial_data.loc[ticker]
E = mkt.get("시가총액", np.nan)
sigma_E = mkt.get("sigma_E", np.nan)
D = fin.get("default_point", np.nan)
if pd.isna(E) or pd.isna(sigma_E) or pd.isna(D) or D <= 0 or E <= 0:
continue
# Merton 풀이
sol = solve_merton(E, sigma_E, D, r, T)
if sol["converged"]:
V = sol["V"]
sigma_V = sol["sigma_V"]
DD = calculate_dd(V, sigma_V, D, r=r, T=T)
EDF = calculate_edf(DD)
method = sol["method"]
elif use_naive_fallback:
naive = naive_dd(E, sigma_E, D, r=r, T=T)
V = naive["V"]
sigma_V = naive["sigma_V"]
DD = naive["DD"]
EDF = naive["EDF"]
method = "naive_fallback"
else:
continue
results.append({
"ticker": ticker,
"E": E,
"sigma_E": sigma_E,
"D": D,
"V": V,
"sigma_V": sigma_V,
"DD": DD,
"EDF": EDF,
"leverage": D / V if V > 0 else np.nan,
"method": method,
})
df = pd.DataFrame(results).set_index("ticker")
print(f"[Merton] DD/EDF 산출 완료: {len(df)}개 종목")
print(f" - fsolve: {(df['method']=='fsolve').sum()}")
print(f" - iterative: {(df['method']=='iterative').sum()}")
print(f" - naive_fallback: {(df['method']=='naive_fallback').sum()}")
return df
# ---- 글로벌 벤치마크 등급별 부도율 ----
GLOBAL_DEFAULT_RATES = {
# Moody's 1983-2023 평균 1년 부도율 (근사치)
"AAA": 0.0000,
"AA+": 0.0002,
"AA": 0.0003,
"AA-": 0.0005,
"A+": 0.0006,
"A": 0.0008,
"A-": 0.0012,
"BBB+": 0.0020,
"BBB": 0.0035,
"BBB-": 0.0070,
"BB+": 0.0100,
"BB": 0.0180,
"BB-": 0.0300,
"B+": 0.0450,
"B": 0.0700,
"B-": 0.1100,
"CCC+": 0.1500,
"CCC": 0.2200,
"CCC-": 0.3000,
}
# DD → 등급 매핑 테이블 (글로벌 벤치마크 기반)
DD_RATING_MAP = [
(6.5, "AAA"),
(6.0, "AA+"),
(5.5, "AA"),
(5.0, "AA-"),
(4.5, "A+"),
(4.0, "A"),
(3.5, "A-"),
(3.2, "BBB+"),
(2.8, "BBB"),
(2.5, "BBB-"),
(2.2, "BB+"),
(1.8, "BB"),
(1.5, "BB-"),
(1.2, "B+"),
(0.9, "B"),
(0.6, "B-"),
(0.3, "CCC+"),
(0.0, "CCC"),
(-999, "CCC-"),
]
def dd_to_rating(dd: float) -> str:
"""DD 값을 신용등급으로 매핑"""
if np.isnan(dd):
return "NR"
for threshold, rating in DD_RATING_MAP:
if dd >= threshold:
return rating
return "CCC-"
def assign_dd_ratings(df: pd.DataFrame, dd_col: str = "DD") -> pd.DataFrame:
"""전 종목에 DD 기반 등급 부여"""
df = df.copy()
df["dd_rating"] = df[dd_col].apply(dd_to_rating)
return df
# ---- CLI 테스트 ----
if __name__ == "__main__":
# 단일 기업 테스트
print("=== Merton 모형 단일 테스트 ===")
# 예시: 시가총액 10조, 변동성 30%, 부도점 5조, 무위험 3.5%
E = 10_000_000_000_000 # 10조
sigma_E = 0.30
D = 5_000_000_000_000 # 5조
r = 0.035
sol = solve_merton(E, sigma_E, D, r)
print(f" V = {sol['V']/1e12:.2f}조")
print(f" σ_V = {sol['sigma_V']:.4f}")
print(f" 수렴: {sol['converged']}, 방법: {sol['method']}")
DD = calculate_dd(sol['V'], sol['sigma_V'], D, r=r)
EDF = calculate_edf(DD)
rating = dd_to_rating(DD)
print(f" DD = {DD:.4f}")
print(f" EDF = {EDF:.6f} ({EDF*100:.4f}%)")
print(f" 내재등급 = {rating}")
# Naive DD 비교
naive = naive_dd(E, sigma_E, D, r=r)
print(f"\n=== Naive DD 비교 ===")
print(f" DD = {naive['DD']:.4f}")
print(f" EDF = {naive['EDF']:.6f} ({naive['EDF']*100:.4f}%)")
print(f" 내재등급 = {dd_to_rating(naive['DD'])}")

68
test_setup.py Normal file
View File

@@ -0,0 +1,68 @@
"""Quick test: Merton model + Vikunja connectivity"""
import sys
sys.path.insert(0, ".")
# Test 1: Merton model
print("=== Test 1: Merton Model ===")
from src.models.merton import solve_merton, calculate_dd, calculate_edf, dd_to_rating
E = 10_000_000_000_000 # 10조
sigma_E = 0.30
D = 5_000_000_000_000 # 5조
r = 0.035
sol = solve_merton(E, sigma_E, D, r)
DD = calculate_dd(sol['V'], sol['sigma_V'], D, r=r)
EDF = calculate_edf(DD)
rating = dd_to_rating(DD)
print(f" V = {sol['V']/1e12:.2f}조")
print(f" sigma_V = {sol['sigma_V']:.4f}")
print(f" DD = {DD:.4f}")
print(f" EDF = {EDF:.6f} ({EDF*100:.4f}%)")
print(f" Rating = {rating}")
print(f" Converged: {sol['converged']}, Method: {sol['method']}")
# Test 2: Vikunja
print("\n=== Test 2: Vikunja Connectivity ===")
try:
import urllib.request
import json
headers = {
"Authorization": "Bearer tk_070f8e0b715e818bb7178c3815ed5389040eddca",
"Content-Type": "application/json",
}
req = urllib.request.Request(
"https://plan.variet.net/api/v1/projects/11/tasks?per_page=5",
headers=headers
)
with urllib.request.urlopen(req) as resp:
tasks = json.loads(resp.read().decode("utf-8"))
print(f" Vikunja connected! {len(tasks)} tasks found")
for t in tasks:
status = "done" if t["done"] else "todo"
print(f" #{t['id']} [{status}] {t['title']}")
except Exception as e:
print(f" Vikunja error: {e}")
# Test 3: pykrx
print("\n=== Test 3: pykrx ===")
try:
from pykrx import stock
tickers = stock.get_market_ticker_list("20260310", market="KOSPI")
print(f" pykrx connected! KOSPI tickers: {len(tickers)}")
except Exception as e:
print(f" pykrx error: {e}")
# Test 4: DART
print("\n=== Test 4: DART API ===")
try:
import OpenDartReader
dart = OpenDartReader("ef6deb100be436aed88051fd4914dbdb58ff2e94")
corp_list = dart.corp_codes
listed = corp_list[corp_list["stock_code"].notna() & (corp_list["stock_code"] != " ")]
print(f" DART connected! Listed companies: {len(listed)}")
except Exception as e:
print(f" DART error: {e}")
print("\n=== All tests complete ===")