종목코드로 DART 재무제표 찾기, 기업 고유번호 연결하는 법

python -m pip install requests pandas pyarrow python-dotenv

DART_KEY=발급받은_40자리_인증키

#!/usr/bin/env python3
"""OpenDART 기업 고유번호를 받아 종목코드 매핑 Parquet로 저장한다."""

from __future__ import annotations

import argparse
from io import BytesIO
import os
from pathlib import Path
import xml.etree.ElementTree as ET
from zipfile import BadZipFile, ZipFile

from dotenv import load_dotenv
import pandas as pd
import requests


ENDPOINT = "https://opendart.fss.or.kr/api/corpCode.xml"
COLUMNS = ("corp_code", "corp_name", "corp_eng_name", "stock_code", "modify_date")


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--output", type=Path, default=Path("data/dart_corp_codes.parquet"))
    return parser.parse_args()


def read_error_message(content: bytes) -> str:
    try:
        root = ET.fromstring(content)
    except ET.ParseError:
        return "예상하지 못한 응답입니다."
    status = root.findtext("status") or "상태 없음"
    message = root.findtext("message") or "메시지 없음"
    return f"OpenDART {status}: {message}"


def download_corp_codes(api_key: str) -> bytes:
    try:
        response = requests.get(
            ENDPOINT,
            params={"crtfc_key": api_key},
            timeout=30,
        )
    except requests.RequestException:
        raise RuntimeError("OpenDART 연결에 실패했습니다.") from None

    if response.status_code != 200:
        raise RuntimeError(f"OpenDART HTTP {response.status_code}")
    return response.content


def parse_corp_codes(content: bytes) -> pd.DataFrame:
    try:
        with ZipFile(BytesIO(content)) as archive:
            xml_content = archive.read("CORPCODE.xml")
    except (BadZipFile, KeyError):
        raise RuntimeError(read_error_message(content)) from None

    root = ET.fromstring(xml_content)
    records = []
    for company in root.findall("list"):
        records.append(
            {
                column: (company.findtext(column) or "").strip()
                for column in COLUMNS
            }
        )

    corp_codes = pd.DataFrame.from_records(records, columns=COLUMNS)
    corp_codes["stock_code"] = corp_codes["stock_code"].replace("", pd.NA)
    corp_codes["modify_date"] = pd.to_datetime(
        corp_codes["modify_date"],
        format="%Y%m%d",
        errors="raise",
    )
    return corp_codes.sort_values("corp_code").reset_index(drop=True)


def validate_corp_codes(corp_codes: pd.DataFrame) -> dict[str, int]:
    listed = corp_codes["stock_code"].notna()
    invalid_stock_code = listed & ~corp_codes["stock_code"].str.fullmatch(
        r"[0-9A-Z]{6}",
        na=False,
    )
    report = {
        "rows": len(corp_codes),
        "listed_rows": int(listed.sum()),
        "duplicate_corp_codes": int(corp_codes["corp_code"].duplicated().sum()),
        "duplicate_stock_codes": int(corp_codes.loc[listed, "stock_code"].duplicated().sum()),
        "missing_corp_names": int(corp_codes["corp_name"].eq("").sum()),
        "invalid_stock_codes": int(invalid_stock_code.sum()),
    }
    error_keys = (
        "duplicate_corp_codes",
        "duplicate_stock_codes",
        "missing_corp_names",
        "invalid_stock_codes",
    )
    if any(report[key] for key in error_keys):
        raise ValueError(f"기업 고유번호 품질 검사 실패: {report}")
    return report


def save_atomically(corp_codes: pd.DataFrame, output: Path) -> None:
    output.parent.mkdir(parents=True, exist_ok=True)
    temporary = output.with_suffix(".tmp.parquet")
    corp_codes.to_parquet(temporary, index=False)
    temporary.replace(output)


def main() -> None:
    args = parse_args()
    load_dotenv()
    api_key = os.getenv("DART_KEY")
    if not api_key:
        raise SystemExit(".env에 DART_KEY를 입력해야 합니다.")

    corp_codes = parse_corp_codes(download_corp_codes(api_key))
    report = validate_corp_codes(corp_codes)
    save_atomically(corp_codes, args.output)

    print(f"전체 회사: {report['rows']:,}")
    print(f"종목코드 보유 회사: {report['listed_rows']:,}")
    print("중복·회사명·종목코드 검사: 정상")
    print(f"저장 파일: {args.output}")


if __name__ == "__main__":
    main()

python opendart_corp_codes.py