DART OpenAPI로 상장사 재무제표(분기 포함)를 파이썬으로 받기: 차단당하지 않는 방법까지

이 블로그의 국내 백테스트(린치 PEG, 미너비니 펀더멘털(연간), 미너비니 펀더멘털(분기))는 모두 금융감독원 전자공시시스템(DART)의 공개 API로 받은 재무 데이터를 썼다. 이 글은 그때 쓴 방법을 정리한 것이다. 분기 단독 금액을 읽는 법, 4분기를 구하는 법, 미래 정보를 막는 법, 그리고 호출을 너무 빨리 해서 접속이 차단됐던 경험과 안전하게 대량으로 받는 방법까지 담았다.

1. 준비: 인증키

• opendart.fss.or.kr에서 회원가입 후 인증키를 신청하면 40자리 키가 이메일로 온다. 이 단계는 본인이 직접 해야 한다.
• 키는 코드에 직접 쓰지 말고 환경변수로 보관한다. 코드를 공유하거나 올릴 때 키가 새는 것을 막기 위해서다. 윈도우에서는 한 번 설정해 두면 새 터미널부터 쓸 수 있다.

setx DART_API_KEY "발급받은_키"

import os, io, time, zipfile, pickle
import xml.etree.ElementTree as ET
import pandas as pd
import requests

KEY = os.environ["DART_API_KEY"]
BASE = "https://opendart.fss.or.kr/api"
REPRT = {"1Q": "11013", "2Q": "11012", "3Q": "11014", "FY": "11011"}   # 1분기, 반기, 3분기, 사업보고서

2. 종목코드를 DART 고유번호로 바꾸기

DART는 종목코드(6자리)가 아니라 고유번호(8자리)로 회사를 부른다. 전체 회사 목록을 한 번 받아서 매핑표를 만든다.

def corp_codes():
    """종목코드 -> DART 고유번호 표 (상장사만)"""
    r = requests.get(f"{BASE}/corpCode.xml", params={"crtfc_key": KEY}, timeout=60)
    z = zipfile.ZipFile(io.BytesIO(r.content))
    root = ET.fromstring(z.read(z.namelist()[0]))
    rows = [(i.findtext("stock_code").strip(), i.findtext("corp_code"), i.findtext("corp_name"))
            for i in root.findall("list") if (i.findtext("stock_code") or "").strip()]
    return pd.DataFrame(rows, columns=["stock_code", "corp_code", "corp_name"]).set_index("stock_code")

cc = corp_codes()          # 상장사 약 4,000개

3. 여러 회사의 재무를 한 번에 받기

회사 하나씩 호출하면 수천 번이 필요하다. 다중회사 주요계정 API(fnlttMultiAcnt)는 고유번호를 최대 100개까지 쉼표로 묶어 한 번에 받을 수 있다.

class Dart:
    def __init__(self, gap=0.65):
        self.gap, self.last, self.s = gap, 0.0, requests.Session()

    def get(self, endpoint, **params):
        wait = self.gap - (time.time() - self.last)         # 호출 사이 간격을 지킨다
        if wait > 0:
            time.sleep(wait)
        self.last = time.time()
        j = self.s.get(f"{BASE}/{endpoint}.json", params={"crtfc_key": KEY, **params}, timeout=30).json()
        if j["status"] == "020":
            raise SystemExit("일일 요청 한도 초과: 내일 이어서 받는다")
        return j                                             # "000" 정상, "013" 해당 자료 없음

def multi_accounts(dart, corp_list, year, reprt):
    """corp_list: 고유번호 리스트(최대 100개)"""
    j = dart.get("fnlttMultiAcnt", corp_code=",".join(corp_list), bsns_year=str(year), reprt_code=REPRT[reprt])
    return pd.DataFrame(j.get("list", []))

응답에서 알아 둘 열은 아래와 같다.

열	의미
fs_div	CFS = 연결재무제표, OFS = 별도재무제표
sj_div	BS = 재무상태표, IS = 손익계산서
account_nm	계정 이름 (매출액, 영업이익, 당기순이익 등)
thstrm_amount	당기 금액. 분기 보고서에서는 그 분기 단독 금액
thstrm_add_amount	당기 누적 금액 (반기, 3분기 보고서에서 의미 있음)
frmtrm_amount	전년 같은 분기 금액
frmtrm_add_amount	전년 같은 기간 누적 금액
rcept_no	접수번호. 앞 8자리가 공시일(YYYYMMDD)

4. 분기 단독 금액과 전년 동기 대비 성장률

분기 보고서 한 건에 이번 분기 단독 금액과 전년 같은 분기 금액이 함께 들어 있어서 별도 계산 없이 전년 동기 대비(YoY) 성장률을 구할 수 있다. 반기 보고서의 "당기" 금액도 상반기 합계가 아니라 2분기 단독 금액이다(누적은 thstrm_add_amount).

def pick(df, account):
    """손익계산서에서 한 계정을 골라 숫자로 바꾼다. 연결이 없으면 별도로 대체."""
    x = df[(df.sj_div == "IS") & (df.account_nm.str.startswith(account))]
    have = x.groupby("corp_code").fs_div.apply(lambda s: "CFS" if (s == "CFS").any() else "OFS")
    x = x[x.fs_div == x.corp_code.map(have)].drop_duplicates("corp_code").copy()
    for c in ("thstrm_amount", "thstrm_add_amount", "frmtrm_amount", "frmtrm_add_amount"):
        if c in x:
            x[c] = pd.to_numeric(x[c].astype(str).str.replace(",", ""), errors="coerce")
    x["공시일"] = pd.to_datetime(x.rcept_no.str[:8])
    return x.set_index("corp_code")

d = Dart()
q1 = pick(multi_accounts(d, ["00126380", "00164779", "00266961"], 2023, "1Q"), "영업이익")
q1["성장률"] = q1.thstrm_amount / q1.frmtrm_amount.where(q1.frmtrm_amount > 0) - 1   # 전년 동기가 흑자일 때만 계산
print(q1[["thstrm_nm", "thstrm_amount", "frmtrm_amount", "성장률", "공시일"]])

실행 결과(예시 회사 3곳, 2023년 1분기 영업이익, 단위: 억 원):

회사	이번 분기	전년 동기	성장률	공시일
A	6,402	141,214	-95.5%	2023-05-15
B	-34,023	28,639	-218.8%	2023-05-15
C	3,305	3,018	+9.5%	2023-05-12

B처럼 흑자에서 적자로 돌아선 경우 성장률이 -100% 아래로 나온다. 이런 값을 그대로 쓰면 해석이 틀어지니, 적자 전환은 따로 분류하는 편이 좋다.

5. 4분기 구하기: 연간 − 3분기 누적

4분기는 별도 보고서가 없다. 사업보고서(연간)에서 3분기 보고서의 누적 금액을 빼서 구한다. 전년 4분기도 같은 방식이다(frmtrm 열을 쓴다).

codes = ["00126380", "00164779", "00266961"]
fy = pick(multi_accounts(d, codes, 2023, "FY"), "영업이익")
q3 = pick(multi_accounts(d, codes, 2023, "3Q"), "영업이익")

q4 = pd.DataFrame({"연간": fy.thstrm_amount, "3분기누적": q3.thstrm_add_amount})
q4["4분기"] = q4["연간"] - q4["3분기누적"]
q4["전년4분기"] = fy.frmtrm_amount - q3.frmtrm_add_amount
print(q4)

회사	연간	3분기 누적	4분기 단독	전년 4분기
A	65,670	37,423	28,247	43,061
B	-77,303	-80,763	3,460	-19,122
C	14,888	10,833	4,055	3,365

단위는 억 원이다. 4분기 단독 값은 회사가 발표한 4분기 실적과 일치하는지 한두 곳은 직접 대조해 보길 권한다.

6. 공시일로 미래 정보 막기

백테스트에서 가장 흔한 실수는 아직 공시되지 않은 실적을 미리 쓰는 것이다. 접수번호 앞 8자리가 공시일이니, 그 다음 날부터 그 재무를 쓰면 된다.

q1["사용시작일"] = q1["공시일"] + pd.Timedelta(days=1)
# 백테스트의 날짜 d 에서는 사용시작일 <= d 인 보고서 중 가장 최근 것만 쓴다

7. 대량으로 안전하게 받기

여러 해, 여러 분기, 수천 개 회사를 받으려면 호출이 수백~수천 건이다. 아래 함수는 간격을 지키고, 실패한 호출은 저장하지 않으며, 중간에 끊겨도 이어받는다.

def fetch_all(dart, corp_list, years, reprts, cache_path):
    raw = pickle.load(open(cache_path, "rb")) if os.path.exists(cache_path) else {}
    batches = [corp_list[i:i + 100] for i in range(0, len(corp_list), 100)]
    for y in years:
        for rp in reprts:
            for bi, b in enumerate(batches):
                if (y, rp, bi) in raw:
                    continue                                   # 이미 받은 것은 건너뛴다
                try:
                    j = dart.get("fnlttMultiAcnt", corp_code=",".join(b), bsns_year=str(y), reprt_code=REPRT[rp])
                except requests.ConnectionError:               # 차단 신호: 더 두드리지 말고 저장 후 중단
                    pickle.dump(raw, open(cache_path, "wb"))
                    print("연결이 끊겨 중단했다. 나중에 다시 실행하면 이어서 받는다.")
                    return raw
                if j["status"] in ("000", "013"):              # 정상, 자료 없음만 저장
                    raw[(y, rp, bi)] = j.get("list", [])
        pickle.dump(raw, open(cache_path, "wb"))
    return raw

직접 겪은 접속 차단

처음에는 이 간격 없이 스레드 8개로 주식수 API를 한꺼번에 호출했다. 1,000건을 몇 초 만에 보냈더니 DART가 연결을 강제로 끊었다. 파이썬에서는 ConnectionResetError(10054)가 났고, 곧 DART 홈페이지까지 브라우저에서 열리지 않았다. 같은 컴퓨터에서 네이버나 구글은 정상이었다. 몇 시간이 지나서야 접속이 풀렸다. 이때 얻은 교훈은 다음과 같다.

문제	대처
짧은 시간에 호출을 몰아서 보냄	호출 간격 0.5초 이상, 스레드를 늘리지 않는다. 분기 재무 627건은 초당 약 1.5건으로 약 11분에 받았고 차단되지 않았다.
연결이 끊겼는데도 계속 재시도	끊기면 즉시 중단하고 저장한다. 더 두드리면 차단이 길어질 수 있다.
실패한 호출을 "자료 없음"으로 저장하는 버그	상태가 000(정상)과 013(자료 없음)일 때만 저장한다. 실패까지 저장하면 나중에 이어받을 때 그 구간이 영원히 빈 채로 남는다.
하루 호출 한도	개인 키는 하루 2만 건 안팎으로 알려져 있다(정확한 한도는 공식 안내를 확인). 초과하면 상태 020이 오니 내일 이어서 받는다.

그 밖의 함정

• 계정 이름이 회사마다 조금씩 다르다. 당기순이익이 "당기순이익", "당기순이익(손실)", 분기에는 "분기순이익" 같은 식이다. startswith나 정규식으로 묶는다.
• 금융회사는 "매출액"이 없다. 영업수익, 이자수익 같은 다른 이름을 쓰니 매출 기준 분석에서는 따로 처리하거나 제외한다.
• 연결과 별도가 섞인다. 같은 회사는 항상 같은 기준으로 비교하고, 연결이 있으면 연결을 우선한다.
• 정정 공시가 있으면 같은 기간의 보고서가 여러 건일 수 있다. 접수번호가 가장 이른 것(최초 공시)을 쓰면 당시 시점에 알 수 있던 값을 쓰게 된다.
• 아직 공시되지 않은 분기를 요청하면 013(자료 없음)이 온다. 오류가 아니다.
• 분기 단독 금액은 회사가 직접 공시하지 않은 4분기만 계산이 필요하다. 1~3분기는 위 방법대로 바로 읽힌다.

이 코드로 한 일

• 피터 린치의 PEG를 국내 주식에서 백테스트: 연간 순이익과 시가총액으로 PER, PEG를 계산
• 미너비니의 펀더멘털 조건(연간 근사), 분기 재무로 다시 검증: 분기 YoY 성장률, 성장 가속, 4분기 계산
• 데이터 오류를 점검하는 체크리스트는 백테스트 데이터 오류 6가지에 정리했다.

공개 API는 모두가 함께 쓰는 서비스다. 필요한 만큼만, 천천히 받는 것이 나를 위해서도 가장 안전한 방법이었다.

---

※ 데이터 출처: 금융감독원 전자공시시스템(DART) OpenAPI. 예제의 회사 A, B, C는 설명을 위한 익명 처리이며 특정 종목을 추천하지 않습니다. 코드는 설명을 위한 예시이며, 실제 사용 시 DART의 이용 안내와 본인의 데이터 구조에 맞게 확인해야 합니다.